Merge pull request #68 from beclab/ci/upload_to_cloud

ci: upload to cloud

Author: aby913

pkg/common/constant.go

@@ -56,7 +56,7 @@ const (
 	Pending   = "pending"
 	Running   = "running"
 	Failed    = "failed"
-	Cancelled = "cancelled"
+	Canceled  = "canceled"
 	Completed = "completed"
 )
 

pkg/common/utils.go

@@ -220,3 +220,29 @@ func EscapeAndJoin(input string, delimiter string) string {
 	}
 	return strings.Join(segments, delimiter)
 }
+
+func PathExists(path string) bool {
+	_, err := os.Stat(path)
+	if err == nil {
+		return true
+	}
+
+	if os.IsNotExist(err) {
+		return false
+	}
+	return false
+}
+
+// ListContains returns a boolean that v is in items
+func ListContains[T comparable](items []T, v T) bool {
+	if items == nil {
+		return false
+	}
+
+	for _, item := range items {
+		if v == item {
+			return true
+		}
+	}
+	return false
+}

pkg/drivers/clouds/cloud.go

@@ -7,9 +7,11 @@ import (
 	"files/pkg/diskcache"
 	"files/pkg/drivers/base"
 	"files/pkg/drivers/clouds/rclone/operations"
+	"files/pkg/drivers/posix/upload"
 	"files/pkg/files"
 	"files/pkg/models"
 	"files/pkg/preview"
+	"files/pkg/tasks"
 	"fmt"
 	"net/url"
 	"path/filepath"
@@ -356,6 +358,8 @@ func (s *CloudStorage) Delete(fileDeleteArg *models.FileDeleteArgs) ([]byte, err
 
 	klog.Infof("Cloud delete, user: %s, param: %s", user, common.ToJson(fileParam))
 
+	// todo need to copy .keep file in fileParam.Path first
+
 	var invalidPaths []string
 
 	for _, dirent := range dirents {
@@ -370,27 +374,7 @@ func (s *CloudStorage) Delete(fileDeleteArg *models.FileDeleteArgs) ([]byte, err
 		return common.ToBytes(invalidPaths), fmt.Errorf("invalid path")
 	}
 
-	for _, dp := range dirents {
-		dp = strings.TrimSpace(dp) //  /path/ or /file
-
-		klog.Infof("Cloud delete, user: %s, dirent: %s", user, dp)
-
-		dpd, err := url.PathUnescape(dp)
-		if err != nil {
-			klog.Errorf("Cloud delete, path unescape error: %v, path: %s", err, dp)
-			deleteFailedPaths = append(deleteFailedPaths, dp)
-			continue
-		}
-
-		_, err = s.service.Delete(fileParam, dpd)
-		if err != nil {
-			klog.Errorf("Cloud delete, delete files error: %v, user: %s", err, user)
-			deleteFailedPaths = append(deleteFailedPaths, dp)
-			continue
-		}
-
-		klog.Infof("Cloud delete, delete success, user: %s, file: %s", user, dpd)
-	}
+	deleteFailedPaths, _ = s.service.command.Delete(fileParam, dirents)
 
 	if err := s.service.command.GetOperation().FsCacheClear(); err != nil {
 		klog.Errorf("Cloud delete, fscache clear error: %v", err)
@@ -539,14 +523,101 @@ func (s *CloudStorage) getFiles(fileParam *models.FileParam) (*models.CloudListR
 	return res, nil
 }
 
+/**
+ * UploadLink
+ */
 func (s *CloudStorage) UploadLink(fileUploadArg *models.FileUploadArgs) ([]byte, error) {
-	return nil, nil
+	var user = fileUploadArg.FileParam.Owner
+
+	klog.Infof("Cloud uploadLink, user: %s, param: %s", user, common.ToJson(fileUploadArg.FileParam))
+
+	data, err := upload.HandleUploadLink(fileUploadArg.FileParam, fileUploadArg.From)
+
+	klog.Infof("Cloud uploadLink, done! data: %s", string(data))
+
+	return data, err
 }
 
+/**
+ * UploadedBytes
+ */
 func (s *CloudStorage) UploadedBytes(fileUploadArg *models.FileUploadArgs) ([]byte, error) {
-	return nil, nil
+	var user = fileUploadArg.FileParam.Owner
+	klog.Infof("Cloud uploadBytes, user: %s, param: %s", user, common.ToJson(fileUploadArg))
+
+	data, err := upload.HandleUploadedBytes(fileUploadArg.FileParam, fileUploadArg.FileName)
+
+	klog.Infof("Cloud uploadBytes, done! data: %s", string(data))
+
+	return data, err
 }
 
+/**
+ * UploadChunks
+ */
 func (s *CloudStorage) UploadChunks(fileUploadArg *models.FileUploadArgs) ([]byte, error) {
-	return nil, nil
+	var user = fileUploadArg.FileParam.Owner
+	var uploadId = fileUploadArg.UploadId
+	var chunkInfo = fileUploadArg.ChunkInfo
+	var taskId = chunkInfo.UploadToCloudTaskId
+	// var canceled = chunkInfo.UploadToCloudTaskCancel
+
+	if taskId != "" { // ~ query task status
+		// todo cancel
+		return s.service.checkUploadTaskState(user, uploadId, taskId, chunkInfo)
+	}
+
+	klog.Infof("Cloud uploadChunks, user: %s, uploadId: %s, param: %s", user, fileUploadArg.UploadId, common.ToJson(fileUploadArg.FileParam))
+
+	ok, fileInfo, err := upload.HandleUploadChunks(fileUploadArg.FileParam, fileUploadArg.UploadId, *fileUploadArg.ChunkInfo, fileUploadArg.Ranges)
+
+	if err != nil {
+		return nil, err
+	}
+
+	if fileInfo == nil {
+		return common.ToBytes(&upload.FileUploadSucced{Success: true}), nil
+	}
+
+	if !ok {
+		return common.ToBytes(fileInfo), nil // frontend ignored
+	}
+
+	klog.Infof("Cloud uploadChunks, phase done, tempPath: %s, data: %s", fileInfo.UploadTempPath, common.ToJson(fileInfo))
+
+	var uploadTempPath = fileInfo.UploadTempPath
+	if !strings.HasSuffix(uploadTempPath, "/") {
+		uploadTempPath = uploadTempPath + "/"
+	}
+	var srcParam = &models.FileParam{}
+	srcParam.GetFileParam(uploadTempPath)
+	srcParam.Path = srcParam.Path + fileInfo.Id
+
+	var dstParam = fileUploadArg.FileParam
+
+	uploadCopyParam, err := s.service.CreateUploadParam(srcParam, dstParam, fileInfo.Name, fileInfo.FileInfo.FileRelativePath)
+	if err != nil {
+		return nil, err
+	}
+
+	klog.Infof("Cloud uploadChunks, uploadToCloud param: %s", common.ToJson(uploadCopyParam))
+
+	var task = tasks.TaskManager.CreateTask(uploadCopyParam)
+	if err = task.Execute(task.UploadToCloud); err != nil {
+		klog.Errorf("Cloud uploadChunks, execute uploadToCloud error: %v", err)
+		return nil, err
+	}
+
+	taskId = task.Id()
+
+	klog.Infof("Cloud uploadChunks, uploadToCloud waiting for execute, taskId: %s", taskId)
+
+	return common.ToBytes(
+		&upload.FileUploadSucced{
+			Success: true,
+			IsCloud: true,
+			TaskId:  taskId,
+			Size:    fileInfo.Size,
+			State:   common.Pending,
+		}), nil
 }

pkg/drivers/clouds/rclone/common/common.go

@@ -4,3 +4,10 @@ var (
 	ServeAddr = "http://127.0.0.1:5572"
 	ServeHost = "127.0.0.1"
 )
+
+type ErrorMessage struct {
+	Error  string      `json:"error"`
+	Input  interface{} `json:"input"`
+	Path   string      `json:"path"`
+	Status int         `json:"status"`
+}

pkg/drivers/clouds/rclone/interface.go

@@ -19,10 +19,11 @@ type Interface interface {
 	GetJob() job.Interface
 
 	GetFilesSize(fileParam *models.FileParam) (int64, error)
-	GetFilesList(param *models.FileParam) (*operations.OperationsList, error)
+	GetFilesList(param *models.FileParam, getPrefix bool) (*operations.OperationsList, error)
 	CreateEmptyDirectories(src, target *models.FileParam) error
 
 	Copy(src, dst *models.FileParam) (*operations.OperationsAsyncJobResp, error)
+	Delete(param *models.FileParam, dirents []string) ([]string, error)
 	Clear(param *models.FileParam) error
 }
 

pkg/drivers/clouds/rclone/operations/operations.go

@@ -265,10 +265,11 @@ func (o *operations) Deletefile(fs string, remote string) error {
 
 func (o *operations) Purge(fs string, remote string) error {
 	var url = fmt.Sprintf("%s/%s", common.ServeAddr, PurgePath)
-
+	var leaveRoot = true
 	var param = OperationsReq{
-		Fs:     fs,     // xxx:yyy/parent
-		Remote: remote, // dir/
+		Fs:        fs,     // xxx:yyy/parent
+		Remote:    remote, // dir/
+		LeaveRoot: &leaveRoot,
 	}
 
 	klog.Infof("[rclone] operations purge, param: %s", commonutils.ToJson(param))
@@ -285,20 +286,20 @@ func (o *operations) Purge(fs string, remote string) error {
 }
 
 func (o *operations) FsCacheClear() error {
-	var url = fmt.Sprintf("%s/%s", common.ServeAddr, FsCacheClearPath)
+	// var url = fmt.Sprintf("%s/%s", common.ServeAddr, FsCacheClearPath)
 
-	klog.Info("[rclone] operations fscacheClear")
+	// klog.Info("[rclone] operations fscacheClear")
 
-	var header = make(http.Header)
-	header.Add("Content-Type", "application/octet-stream")
+	// var header = make(http.Header)
+	// header.Add("Content-Type", "application/octet-stream")
 
-	_, err := utils.Request(context.Background(), url, http.MethodPost, &header, nil)
-	if err != nil {
-		klog.Errorf("[rclone] operations fscacheClear error: %v", err)
-		return err
-	}
+	// _, err := utils.Request(context.Background(), url, http.MethodPost, &header, nil)
+	// if err != nil {
+	// 	klog.Errorf("[rclone] operations fscacheClear error: %v", err)
+	// 	return err
+	// }
 
-	klog.Info("[rclone] operations fscacheClear done!")
+	// klog.Info("[rclone] operations fscacheClear done!")
 
 	return nil
 }

pkg/drivers/clouds/rclone/rclone.go

@@ -10,6 +10,7 @@ import (
 	"files/pkg/files"
 	"files/pkg/models"
 	"fmt"
+	"net/url"
 	"path/filepath"
 	"sort"
 	"strings"
@@ -258,7 +259,7 @@ func (r *rclone) GetFilesSize(fileParam *models.FileParam) (int64, error) {
 
 }
 
-func (r *rclone) GetFilesList(param *models.FileParam) (*operations.OperationsList, error) {
+func (r *rclone) GetFilesList(param *models.FileParam, getPrefix bool) (*operations.OperationsList, error) {
 
 	var fsPrefix, err = r.GetFsPrefix(param)
 	if err != nil {
@@ -271,7 +272,13 @@ func (r *rclone) GetFilesList(param *models.FileParam) (*operations.OperationsLi
 		Metadata:   false,
 	}
 
-	var fs = fsPrefix + pathPrefix
+	var fs string
+	if getPrefix {
+		fs = fsPrefix + pathPrefix
+	} else {
+		fs = fsPrefix + param.Path
+	}
+
 	lists, err := r.operation.List(fs, opt)
 	if err != nil {
 		return nil, err
@@ -430,6 +437,7 @@ func (r *rclone) Clear(param *models.FileParam) error {
 	var isSrcLocal bool
 	var owner = param.Owner
 
+	klog.Infof("[rclone] clear, param: %s", common.ToJson(param))
 	fileName, isFile := files.GetFileNameFromPath(param.Path)
 	prefixPath := files.GetPrefixPath(param.Path)
 
@@ -465,13 +473,13 @@ func (r *rclone) Clear(param *models.FileParam) error {
 		remote = fileName
 
 		if err = r.GetOperation().Deletefile(fs, remote); err != nil {
-			klog.Errorf("[rclone] clear file error: %v, user: %s, fs: %s, remote: %s", err, owner, fs, remote)
+			klog.Errorf("[rclone] clear, delete file error: %v, user: %s, isFile: %v, fs: %s, remote: %s", err, owner, isFile, fs, remote)
 			return err
 		}
 
 		r.GetOperation().FsCacheClear()
 
-		klog.Infof("[rclone] clear file done! user: %s, fs: %s, remote: %s", owner, fs, remote)
+		klog.Infof("[rclone] clear, file done! user: %s, fs: %s, remote: %s", owner, fs, remote)
 
 		return nil
 	}
@@ -481,13 +489,60 @@ func (r *rclone) Clear(param *models.FileParam) error {
 	var remote = fileName
 
 	if err = r.GetOperation().Purge(fs, remote); err != nil {
-		klog.Errorf("[rclone] clear directory error: %v, user: %s, fs: %s, remote: %s", err, owner, fs, remote)
+		klog.Errorf("[rclone] clear, purge error: %v, user: %s, fs: %s, remote: %s", err, owner, fs, remote)
 		return err
 	}
 
 	r.GetOperation().FsCacheClear()
 
-	klog.Infof("[rclone] clear directory done! user: %s, fs: %s, remote: %s", owner, fs, remote)
+	klog.Infof("[rclone] clear, purge done! user: %s, fs: %s, remote: %s", owner, fs, remote)
 
 	return nil
 }
+
+func (r *rclone) Delete(param *models.FileParam, dirents []string) ([]string, error) {
+	var user = param.Owner
+	var deleteFailedPaths []string
+	var total = len(dirents)
+
+	for current, dp := range dirents {
+		dp = strings.TrimSpace(dp) //  /path/ or /file
+
+		dpd, err := url.PathUnescape(dp)
+		if err != nil {
+			klog.Errorf("[rclone] delete, path unescape error: %v, path: %s", err, dp)
+			deleteFailedPaths = append(deleteFailedPaths, dp)
+			continue
+		}
+
+		klog.Infof("[rclone] delete, delete (%d/%d), user: %s, file: %s", current+1, total, user, dpd)
+
+		fsPrefix, err := r.GetFsPrefix(param)
+		_, isFile := files.GetFileNameFromPath(dpd)
+
+		var fs, remote string
+
+		if isFile {
+			fs = fsPrefix + param.Path
+			remote = strings.TrimPrefix(dpd, "/")
+			if err = r.GetOperation().Deletefile(fs, remote); err != nil {
+				deleteFailedPaths = append(deleteFailedPaths, dp)
+			}
+
+		} else {
+			fs = fsPrefix + param.Path
+			remote = strings.TrimPrefix(dpd, "/")
+
+			if err = r.GetOperation().Purge(fs, remote); err != nil {
+				deleteFailedPaths = append(deleteFailedPaths, dp)
+			}
+		}
+	}
+
+	if len(deleteFailedPaths) > 0 {
+		return deleteFailedPaths, fmt.Errorf("delete failed paths")
+	}
+
+	return nil, nil
+
+}