e2f47ecdeb
See #5538
1219 lines
33 KiB
Go
1219 lines
33 KiB
Go
// Package mega provides an interface to the Mega
|
|
// object storage system.
|
|
package mega
|
|
|
|
/*
|
|
Open questions
|
|
* Does mega support a content hash - what exactly are the mega hashes?
|
|
* Can mega support setting modification times?
|
|
|
|
Improvements:
|
|
* Uploads could be done in parallel
|
|
* Downloads would be more efficient done in one go
|
|
* Uploads would be more efficient with bigger chunks
|
|
* Looks like mega can support server-side copy, but it isn't implemented in go-mega
|
|
* Upload can set modtime... - set as int64_t - can set ctime and mtime?
|
|
*/
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"path"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/pkg/errors"
|
|
"github.com/rclone/rclone/fs"
|
|
"github.com/rclone/rclone/fs/config"
|
|
"github.com/rclone/rclone/fs/config/configmap"
|
|
"github.com/rclone/rclone/fs/config/configstruct"
|
|
"github.com/rclone/rclone/fs/config/obscure"
|
|
"github.com/rclone/rclone/fs/fserrors"
|
|
"github.com/rclone/rclone/fs/fshttp"
|
|
"github.com/rclone/rclone/fs/hash"
|
|
"github.com/rclone/rclone/lib/encoder"
|
|
"github.com/rclone/rclone/lib/pacer"
|
|
"github.com/rclone/rclone/lib/readers"
|
|
mega "github.com/t3rm1n4l/go-mega"
|
|
)
|
|
|
|
const (
|
|
minSleep = 10 * time.Millisecond
|
|
maxSleep = 2 * time.Second
|
|
eventWaitTime = 500 * time.Millisecond
|
|
decayConstant = 2 // bigger for slower decay, exponential
|
|
)
|
|
|
|
var (
|
|
megaCacheMu sync.Mutex // mutex for the below
|
|
megaCache = map[string]*mega.Mega{} // cache logged in Mega's by user
|
|
)
|
|
|
|
// Register with Fs
|
|
func init() {
|
|
fs.Register(&fs.RegInfo{
|
|
Name: "mega",
|
|
Description: "Mega",
|
|
NewFs: NewFs,
|
|
Options: []fs.Option{{
|
|
Name: "user",
|
|
Help: "User name.",
|
|
Required: true,
|
|
}, {
|
|
Name: "pass",
|
|
Help: "Password.",
|
|
Required: true,
|
|
IsPassword: true,
|
|
}, {
|
|
Name: "debug",
|
|
Help: `Output more debug from Mega.
|
|
|
|
If this flag is set (along with -vv) it will print further debugging
|
|
information from the mega backend.`,
|
|
Default: false,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "hard_delete",
|
|
Help: `Delete files permanently rather than putting them into the trash.
|
|
|
|
Normally the mega backend will put all deletions into the trash rather
|
|
than permanently deleting them. If you specify this then rclone will
|
|
permanently delete objects instead.`,
|
|
Default: false,
|
|
Advanced: true,
|
|
}, {
|
|
Name: config.ConfigEncoding,
|
|
Help: config.ConfigEncodingHelp,
|
|
Advanced: true,
|
|
// Encode invalid UTF-8 bytes as json doesn't handle them properly.
|
|
Default: (encoder.Base |
|
|
encoder.EncodeInvalidUtf8),
|
|
}},
|
|
})
|
|
}
|
|
|
|
// Options defines the configuration for this backend
|
|
type Options struct {
|
|
User string `config:"user"`
|
|
Pass string `config:"pass"`
|
|
Debug bool `config:"debug"`
|
|
HardDelete bool `config:"hard_delete"`
|
|
Enc encoder.MultiEncoder `config:"encoding"`
|
|
}
|
|
|
|
// Fs represents a remote mega
|
|
type Fs struct {
|
|
name string // name of this remote
|
|
root string // the path we are working on
|
|
opt Options // parsed config options
|
|
features *fs.Features // optional features
|
|
srv *mega.Mega // the connection to the server
|
|
pacer *fs.Pacer // pacer for API calls
|
|
rootNodeMu sync.Mutex // mutex for _rootNode
|
|
_rootNode *mega.Node // root node - call findRoot to use this
|
|
mkdirMu sync.Mutex // used to serialize calls to mkdir / rmdir
|
|
}
|
|
|
|
// Object describes a mega object
|
|
//
|
|
// Will definitely have info but maybe not meta
|
|
//
|
|
// Normally rclone would just store an ID here but go-mega and mega.nz
|
|
// expect you to build an entire tree of all the objects in memory.
|
|
// In this case we just store a pointer to the object.
|
|
type Object struct {
|
|
fs *Fs // what this object is part of
|
|
remote string // The remote path
|
|
info *mega.Node // pointer to the mega node
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// Name of the remote (as passed into NewFs)
|
|
func (f *Fs) Name() string {
|
|
return f.name
|
|
}
|
|
|
|
// Root of the remote (as passed into NewFs)
|
|
func (f *Fs) Root() string {
|
|
return f.root
|
|
}
|
|
|
|
// String converts this Fs to a string
|
|
func (f *Fs) String() string {
|
|
return fmt.Sprintf("mega root '%s'", f.root)
|
|
}
|
|
|
|
// Features returns the optional features of this Fs
|
|
func (f *Fs) Features() *fs.Features {
|
|
return f.features
|
|
}
|
|
|
|
// parsePath parses a mega 'url'
|
|
func parsePath(path string) (root string) {
|
|
root = strings.Trim(path, "/")
|
|
return
|
|
}
|
|
|
|
// shouldRetry returns a boolean as to whether this err deserves to be
|
|
// retried. It returns the err as a convenience
|
|
func shouldRetry(ctx context.Context, err error) (bool, error) {
|
|
if fserrors.ContextError(ctx, &err) {
|
|
return false, err
|
|
}
|
|
// Let the mega library handle the low level retries
|
|
return false, err
|
|
/*
|
|
switch errors.Cause(err) {
|
|
case mega.EAGAIN, mega.ERATELIMIT, mega.ETEMPUNAVAIL:
|
|
return true, err
|
|
}
|
|
return fserrors.ShouldRetry(err), err
|
|
*/
|
|
}
|
|
|
|
// readMetaDataForPath reads the metadata from the path
|
|
func (f *Fs) readMetaDataForPath(ctx context.Context, remote string) (info *mega.Node, err error) {
|
|
rootNode, err := f.findRoot(ctx, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return f.findObject(rootNode, remote)
|
|
}
|
|
|
|
// NewFs constructs an Fs from the path, container:path
|
|
func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, error) {
|
|
// Parse config into Options struct
|
|
opt := new(Options)
|
|
err := configstruct.Set(m, opt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if opt.Pass != "" {
|
|
var err error
|
|
opt.Pass, err = obscure.Reveal(opt.Pass)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "couldn't decrypt password")
|
|
}
|
|
}
|
|
ci := fs.GetConfig(ctx)
|
|
|
|
// cache *mega.Mega on username so we can re-use and share
|
|
// them between remotes. They are expensive to make as they
|
|
// contain all the objects and sharing the objects makes the
|
|
// move code easier as we don't have to worry about mixing
|
|
// them up between different remotes.
|
|
megaCacheMu.Lock()
|
|
defer megaCacheMu.Unlock()
|
|
srv := megaCache[opt.User]
|
|
if srv == nil {
|
|
srv = mega.New().SetClient(fshttp.NewClient(ctx))
|
|
srv.SetRetries(ci.LowLevelRetries) // let mega do the low level retries
|
|
srv.SetLogger(func(format string, v ...interface{}) {
|
|
fs.Infof("*go-mega*", format, v...)
|
|
})
|
|
if opt.Debug {
|
|
srv.SetDebugger(func(format string, v ...interface{}) {
|
|
fs.Debugf("*go-mega*", format, v...)
|
|
})
|
|
}
|
|
|
|
err := srv.Login(opt.User, opt.Pass)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "couldn't login")
|
|
}
|
|
megaCache[opt.User] = srv
|
|
}
|
|
|
|
root = parsePath(root)
|
|
f := &Fs{
|
|
name: name,
|
|
root: root,
|
|
opt: *opt,
|
|
srv: srv,
|
|
pacer: fs.NewPacer(ctx, pacer.NewDefault(pacer.MinSleep(minSleep), pacer.MaxSleep(maxSleep), pacer.DecayConstant(decayConstant))),
|
|
}
|
|
f.features = (&fs.Features{
|
|
DuplicateFiles: true,
|
|
CanHaveEmptyDirectories: true,
|
|
}).Fill(ctx, f)
|
|
|
|
// Find the root node and check if it is a file or not
|
|
_, err = f.findRoot(ctx, false)
|
|
switch err {
|
|
case nil:
|
|
// root node found and is a directory
|
|
case fs.ErrorDirNotFound:
|
|
// root node not found, so can't be a file
|
|
case fs.ErrorIsFile:
|
|
// root node is a file so point to parent directory
|
|
root = path.Dir(root)
|
|
if root == "." {
|
|
root = ""
|
|
}
|
|
f.root = root
|
|
return f, err
|
|
}
|
|
return f, nil
|
|
}
|
|
|
|
// splitNodePath splits nodePath into / separated parts, returning nil if it
|
|
// should refer to the root.
|
|
// It also encodes the parts into backend specific encoding
|
|
func (f *Fs) splitNodePath(nodePath string) (parts []string) {
|
|
nodePath = path.Clean(nodePath)
|
|
if nodePath == "." || nodePath == "/" {
|
|
return nil
|
|
}
|
|
nodePath = f.opt.Enc.FromStandardPath(nodePath)
|
|
return strings.Split(nodePath, "/")
|
|
}
|
|
|
|
// findNode looks up the node for the path of the name given from the root given
|
|
//
|
|
// It returns mega.ENOENT if it wasn't found
|
|
func (f *Fs) findNode(rootNode *mega.Node, nodePath string) (*mega.Node, error) {
|
|
parts := f.splitNodePath(nodePath)
|
|
if parts == nil {
|
|
return rootNode, nil
|
|
}
|
|
nodes, err := f.srv.FS.PathLookup(rootNode, parts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return nodes[len(nodes)-1], nil
|
|
}
|
|
|
|
// findDir finds the directory rooted from the node passed in
|
|
func (f *Fs) findDir(rootNode *mega.Node, dir string) (node *mega.Node, err error) {
|
|
node, err = f.findNode(rootNode, dir)
|
|
if err == mega.ENOENT {
|
|
return nil, fs.ErrorDirNotFound
|
|
} else if err == nil && node.GetType() == mega.FILE {
|
|
return nil, fs.ErrorIsFile
|
|
}
|
|
return node, err
|
|
}
|
|
|
|
// findObject looks up the node for the object of the name given
|
|
func (f *Fs) findObject(rootNode *mega.Node, file string) (node *mega.Node, err error) {
|
|
node, err = f.findNode(rootNode, file)
|
|
if err == mega.ENOENT {
|
|
return nil, fs.ErrorObjectNotFound
|
|
} else if err == nil && node.GetType() != mega.FILE {
|
|
return nil, fs.ErrorIsDir // all other node types are directories
|
|
}
|
|
return node, err
|
|
}
|
|
|
|
// lookupDir looks up the node for the directory of the name given
|
|
//
|
|
// if create is true it tries to create the root directory if not found
|
|
func (f *Fs) lookupDir(ctx context.Context, dir string) (*mega.Node, error) {
|
|
rootNode, err := f.findRoot(ctx, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return f.findDir(rootNode, dir)
|
|
}
|
|
|
|
// lookupParentDir finds the parent node for the remote passed in
|
|
func (f *Fs) lookupParentDir(ctx context.Context, remote string) (dirNode *mega.Node, leaf string, err error) {
|
|
parent, leaf := path.Split(remote)
|
|
dirNode, err = f.lookupDir(ctx, parent)
|
|
return dirNode, leaf, err
|
|
}
|
|
|
|
// mkdir makes the directory and any parent directories for the
|
|
// directory of the name given
|
|
func (f *Fs) mkdir(ctx context.Context, rootNode *mega.Node, dir string) (node *mega.Node, err error) {
|
|
f.mkdirMu.Lock()
|
|
defer f.mkdirMu.Unlock()
|
|
|
|
parts := f.splitNodePath(dir)
|
|
if parts == nil {
|
|
return rootNode, nil
|
|
}
|
|
var i int
|
|
// look up until we find a directory which exists
|
|
for i = 0; i <= len(parts); i++ {
|
|
var nodes []*mega.Node
|
|
nodes, err = f.srv.FS.PathLookup(rootNode, parts[:len(parts)-i])
|
|
if err == nil {
|
|
if len(nodes) == 0 {
|
|
node = rootNode
|
|
} else {
|
|
node = nodes[len(nodes)-1]
|
|
}
|
|
break
|
|
}
|
|
if err != mega.ENOENT {
|
|
return nil, errors.Wrap(err, "mkdir lookup failed")
|
|
}
|
|
}
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "internal error: mkdir called with non existent root node")
|
|
}
|
|
// i is number of directories to create (may be 0)
|
|
// node is directory to create them from
|
|
for _, name := range parts[len(parts)-i:] {
|
|
// create directory called name in node
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
node, err = f.srv.CreateDir(name, node)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "mkdir create node failed")
|
|
}
|
|
}
|
|
return node, nil
|
|
}
|
|
|
|
// mkdirParent creates the parent directory of remote
|
|
func (f *Fs) mkdirParent(ctx context.Context, remote string) (dirNode *mega.Node, leaf string, err error) {
|
|
rootNode, err := f.findRoot(ctx, true)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
parent, leaf := path.Split(remote)
|
|
dirNode, err = f.mkdir(ctx, rootNode, parent)
|
|
return dirNode, leaf, err
|
|
}
|
|
|
|
// findRoot looks up the root directory node and returns it.
|
|
//
|
|
// if create is true it tries to create the root directory if not found
|
|
func (f *Fs) findRoot(ctx context.Context, create bool) (*mega.Node, error) {
|
|
f.rootNodeMu.Lock()
|
|
defer f.rootNodeMu.Unlock()
|
|
|
|
// Check if we haven't found it already
|
|
if f._rootNode != nil {
|
|
return f._rootNode, nil
|
|
}
|
|
|
|
// Check for pre-existing root
|
|
absRoot := f.srv.FS.GetRoot()
|
|
node, err := f.findDir(absRoot, f.root)
|
|
//log.Printf("findRoot findDir %p %v", node, err)
|
|
if err == nil {
|
|
f._rootNode = node
|
|
return node, nil
|
|
}
|
|
if !create || err != fs.ErrorDirNotFound {
|
|
return nil, err
|
|
}
|
|
|
|
//..not found so create the root directory
|
|
f._rootNode, err = f.mkdir(ctx, absRoot, f.root)
|
|
return f._rootNode, err
|
|
}
|
|
|
|
// clearRoot unsets the root directory
|
|
func (f *Fs) clearRoot() {
|
|
f.rootNodeMu.Lock()
|
|
f._rootNode = nil
|
|
f.rootNodeMu.Unlock()
|
|
//log.Printf("cleared root directory")
|
|
}
|
|
|
|
// CleanUp deletes all files currently in trash
|
|
func (f *Fs) CleanUp(ctx context.Context) (err error) {
|
|
trash := f.srv.FS.GetTrash()
|
|
items := []*mega.Node{}
|
|
_, err = f.list(ctx, trash, func(item *mega.Node) bool {
|
|
items = append(items, item)
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "CleanUp failed to list items in trash")
|
|
}
|
|
fs.Infof(f, "Deleting %d items from the trash", len(items))
|
|
errors := 0
|
|
// similar to f.deleteNode(trash) but with HardDelete as true
|
|
for _, item := range items {
|
|
fs.Debugf(f, "Deleting trash %q", f.opt.Enc.ToStandardName(item.GetName()))
|
|
deleteErr := f.pacer.Call(func() (bool, error) {
|
|
err := f.srv.Delete(item, true)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if deleteErr != nil {
|
|
err = deleteErr
|
|
errors++
|
|
}
|
|
}
|
|
fs.Infof(f, "Deleted %d items from the trash with %d errors", len(items), errors)
|
|
return err
|
|
}
|
|
|
|
// Return an Object from a path
|
|
//
|
|
// If it can't be found it returns the error fs.ErrorObjectNotFound.
|
|
func (f *Fs) newObjectWithInfo(ctx context.Context, remote string, info *mega.Node) (fs.Object, error) {
|
|
o := &Object{
|
|
fs: f,
|
|
remote: remote,
|
|
}
|
|
var err error
|
|
if info != nil {
|
|
// Set info
|
|
err = o.setMetaData(info)
|
|
} else {
|
|
err = o.readMetaData(ctx) // reads info and meta, returning an error
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return o, nil
|
|
}
|
|
|
|
// NewObject finds the Object at remote. If it can't be found
|
|
// it returns the error fs.ErrorObjectNotFound.
|
|
func (f *Fs) NewObject(ctx context.Context, remote string) (fs.Object, error) {
|
|
return f.newObjectWithInfo(ctx, remote, nil)
|
|
}
|
|
|
|
// list the objects into the function supplied
|
|
//
|
|
// If directories is set it only sends directories
|
|
// User function to process a File item from listAll
|
|
//
|
|
// Should return true to finish processing
|
|
type listFn func(*mega.Node) bool
|
|
|
|
// Lists the directory required calling the user function on each item found
|
|
//
|
|
// If the user fn ever returns true then it early exits with found = true
|
|
func (f *Fs) list(ctx context.Context, dir *mega.Node, fn listFn) (found bool, err error) {
|
|
nodes, err := f.srv.FS.GetChildren(dir)
|
|
if err != nil {
|
|
return false, errors.Wrapf(err, "list failed")
|
|
}
|
|
for _, item := range nodes {
|
|
if fn(item) {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
// List the objects and directories in dir into entries. The
|
|
// entries can be returned in any order but should be for a
|
|
// complete directory.
|
|
//
|
|
// dir should be "" to list the root, and should not have
|
|
// trailing slashes.
|
|
//
|
|
// This should return ErrDirNotFound if the directory isn't
|
|
// found.
|
|
func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err error) {
|
|
dirNode, err := f.lookupDir(ctx, dir)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var iErr error
|
|
_, err = f.list(ctx, dirNode, func(info *mega.Node) bool {
|
|
remote := path.Join(dir, f.opt.Enc.ToStandardName(info.GetName()))
|
|
switch info.GetType() {
|
|
case mega.FOLDER, mega.ROOT, mega.INBOX, mega.TRASH:
|
|
d := fs.NewDir(remote, info.GetTimeStamp()).SetID(info.GetHash())
|
|
entries = append(entries, d)
|
|
case mega.FILE:
|
|
o, err := f.newObjectWithInfo(ctx, remote, info)
|
|
if err != nil {
|
|
iErr = err
|
|
return true
|
|
}
|
|
entries = append(entries, o)
|
|
}
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if iErr != nil {
|
|
return nil, iErr
|
|
}
|
|
return entries, nil
|
|
}
|
|
|
|
// Creates from the parameters passed in a half finished Object which
|
|
// must have setMetaData called on it
|
|
//
|
|
// Returns the dirNode, object, leaf and error
|
|
//
|
|
// Used to create new objects
|
|
func (f *Fs) createObject(ctx context.Context, remote string, modTime time.Time, size int64) (o *Object, dirNode *mega.Node, leaf string, err error) {
|
|
dirNode, leaf, err = f.mkdirParent(ctx, remote)
|
|
if err != nil {
|
|
return nil, nil, leaf, err
|
|
}
|
|
// Temporary Object under construction
|
|
o = &Object{
|
|
fs: f,
|
|
remote: remote,
|
|
}
|
|
return o, dirNode, leaf, nil
|
|
}
|
|
|
|
// Put the object
|
|
//
|
|
// Copy the reader in to the new object which is returned
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
// PutUnchecked uploads the object
|
|
//
|
|
// This will create a duplicate if we upload a new file without
|
|
// checking to see if there is one already - use Put() for that.
|
|
func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
|
existingObj, err := f.newObjectWithInfo(ctx, src.Remote(), nil)
|
|
switch err {
|
|
case nil:
|
|
return existingObj, existingObj.Update(ctx, in, src, options...)
|
|
case fs.ErrorObjectNotFound:
|
|
// Not found so create it
|
|
return f.PutUnchecked(ctx, in, src)
|
|
default:
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// PutUnchecked the object
|
|
//
|
|
// Copy the reader in to the new object which is returned
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
// PutUnchecked uploads the object
|
|
//
|
|
// This will create a duplicate if we upload a new file without
|
|
// checking to see if there is one already - use Put() for that.
|
|
func (f *Fs) PutUnchecked(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
|
remote := src.Remote()
|
|
size := src.Size()
|
|
modTime := src.ModTime(ctx)
|
|
|
|
o, _, _, err := f.createObject(ctx, remote, modTime, size)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return o, o.Update(ctx, in, src, options...)
|
|
}
|
|
|
|
// Mkdir creates the directory if it doesn't exist
|
|
func (f *Fs) Mkdir(ctx context.Context, dir string) error {
|
|
rootNode, err := f.findRoot(ctx, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = f.mkdir(ctx, rootNode, dir)
|
|
return errors.Wrap(err, "Mkdir failed")
|
|
}
|
|
|
|
// deleteNode removes a file or directory, observing useTrash
|
|
func (f *Fs) deleteNode(ctx context.Context, node *mega.Node) (err error) {
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
err = f.srv.Delete(node, f.opt.HardDelete)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
return err
|
|
}
|
|
|
|
// purgeCheck removes the directory dir, if check is set then it
|
|
// refuses to do so if it has anything in
|
|
func (f *Fs) purgeCheck(ctx context.Context, dir string, check bool) error {
|
|
f.mkdirMu.Lock()
|
|
defer f.mkdirMu.Unlock()
|
|
|
|
rootNode, err := f.findRoot(ctx, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
dirNode, err := f.findDir(rootNode, dir)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if check {
|
|
children, err := f.srv.FS.GetChildren(dirNode)
|
|
if err != nil {
|
|
return errors.Wrap(err, "purgeCheck GetChildren failed")
|
|
}
|
|
if len(children) > 0 {
|
|
return fs.ErrorDirectoryNotEmpty
|
|
}
|
|
}
|
|
|
|
waitEvent := f.srv.WaitEventsStart()
|
|
|
|
err = f.deleteNode(ctx, dirNode)
|
|
if err != nil {
|
|
return errors.Wrap(err, "delete directory node failed")
|
|
}
|
|
|
|
// Remove the root node if we just deleted it
|
|
if dirNode == rootNode {
|
|
f.clearRoot()
|
|
}
|
|
|
|
f.srv.WaitEvents(waitEvent, eventWaitTime)
|
|
return nil
|
|
}
|
|
|
|
// Rmdir deletes the root folder
|
|
//
|
|
// Returns an error if it isn't empty
|
|
func (f *Fs) Rmdir(ctx context.Context, dir string) error {
|
|
return f.purgeCheck(ctx, dir, true)
|
|
}
|
|
|
|
// Precision return the precision of this Fs
|
|
func (f *Fs) Precision() time.Duration {
|
|
return fs.ModTimeNotSupported
|
|
}
|
|
|
|
// Purge deletes all the files in the directory
|
|
//
|
|
// Optional interface: Only implement this if you have a way of
|
|
// deleting all the files quicker than just running Remove() on the
|
|
// result of List()
|
|
func (f *Fs) Purge(ctx context.Context, dir string) error {
|
|
return f.purgeCheck(ctx, dir, false)
|
|
}
|
|
|
|
// move a file or folder (srcFs, srcRemote, info) to (f, dstRemote)
|
|
//
|
|
// info will be updates
|
|
func (f *Fs) move(ctx context.Context, dstRemote string, srcFs *Fs, srcRemote string, info *mega.Node) (err error) {
|
|
var (
|
|
dstFs = f
|
|
srcDirNode, dstDirNode *mega.Node
|
|
srcParent, dstParent string
|
|
srcLeaf, dstLeaf string
|
|
)
|
|
|
|
if dstRemote != "" {
|
|
// lookup or create the destination parent directory
|
|
dstDirNode, dstLeaf, err = dstFs.mkdirParent(ctx, dstRemote)
|
|
} else {
|
|
// find or create the parent of the root directory
|
|
absRoot := dstFs.srv.FS.GetRoot()
|
|
dstParent, dstLeaf = path.Split(dstFs.root)
|
|
dstDirNode, err = dstFs.mkdir(ctx, absRoot, dstParent)
|
|
}
|
|
if err != nil {
|
|
return errors.Wrap(err, "server-side move failed to make dst parent dir")
|
|
}
|
|
|
|
if srcRemote != "" {
|
|
// lookup the existing parent directory
|
|
srcDirNode, srcLeaf, err = srcFs.lookupParentDir(ctx, srcRemote)
|
|
} else {
|
|
// lookup the existing root parent
|
|
absRoot := srcFs.srv.FS.GetRoot()
|
|
srcParent, srcLeaf = path.Split(srcFs.root)
|
|
srcDirNode, err = f.findDir(absRoot, srcParent)
|
|
}
|
|
if err != nil {
|
|
return errors.Wrap(err, "server-side move failed to lookup src parent dir")
|
|
}
|
|
|
|
// move the object into its new directory if required
|
|
if srcDirNode != dstDirNode && srcDirNode.GetHash() != dstDirNode.GetHash() {
|
|
//log.Printf("move src %p %q dst %p %q", srcDirNode, srcDirNode.GetName(), dstDirNode, dstDirNode.GetName())
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
err = f.srv.Move(info, dstDirNode)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "server-side move failed")
|
|
}
|
|
}
|
|
|
|
waitEvent := f.srv.WaitEventsStart()
|
|
|
|
// rename the object if required
|
|
if srcLeaf != dstLeaf {
|
|
//log.Printf("rename %q to %q", srcLeaf, dstLeaf)
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
err = f.srv.Rename(info, f.opt.Enc.FromStandardName(dstLeaf))
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "server-side rename failed")
|
|
}
|
|
}
|
|
|
|
f.srv.WaitEvents(waitEvent, eventWaitTime)
|
|
|
|
return nil
|
|
}
|
|
|
|
// Move src to this remote using server-side move operations.
|
|
//
|
|
// This is stored with the remote path given
|
|
//
|
|
// It returns the destination Object and a possible error
|
|
//
|
|
// Will only be called if src.Fs().Name() == f.Name()
|
|
//
|
|
// If it isn't possible then return fs.ErrorCantMove
|
|
func (f *Fs) Move(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
|
|
dstFs := f
|
|
|
|
//log.Printf("Move %q -> %q", src.Remote(), remote)
|
|
srcObj, ok := src.(*Object)
|
|
if !ok {
|
|
fs.Debugf(src, "Can't move - not same remote type")
|
|
return nil, fs.ErrorCantMove
|
|
}
|
|
|
|
// Do the move
|
|
err := f.move(ctx, remote, srcObj.fs, srcObj.remote, srcObj.info)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Create a destination object
|
|
dstObj := &Object{
|
|
fs: dstFs,
|
|
remote: remote,
|
|
info: srcObj.info,
|
|
}
|
|
return dstObj, nil
|
|
}
|
|
|
|
// DirMove moves src, srcRemote to this remote at dstRemote
|
|
// using server-side move operations.
|
|
//
|
|
// Will only be called if src.Fs().Name() == f.Name()
|
|
//
|
|
// If it isn't possible then return fs.ErrorCantDirMove
|
|
//
|
|
// If destination exists then return fs.ErrorDirExists
|
|
func (f *Fs) DirMove(ctx context.Context, src fs.Fs, srcRemote, dstRemote string) error {
|
|
dstFs := f
|
|
srcFs, ok := src.(*Fs)
|
|
if !ok {
|
|
fs.Debugf(srcFs, "Can't move directory - not same remote type")
|
|
return fs.ErrorCantDirMove
|
|
}
|
|
|
|
// find the source
|
|
info, err := srcFs.lookupDir(ctx, srcRemote)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// check the destination doesn't exist
|
|
_, err = dstFs.lookupDir(ctx, dstRemote)
|
|
if err == nil {
|
|
return fs.ErrorDirExists
|
|
} else if err != fs.ErrorDirNotFound {
|
|
return errors.Wrap(err, "DirMove error while checking dest directory")
|
|
}
|
|
|
|
// Do the move
|
|
err = f.move(ctx, dstRemote, srcFs, srcRemote, info)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Clear src if it was the root
|
|
if srcRemote == "" {
|
|
srcFs.clearRoot()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// DirCacheFlush an optional interface to flush internal directory cache
|
|
func (f *Fs) DirCacheFlush() {
|
|
// f.dirCache.ResetRoot()
|
|
// FIXME Flush the mega somehow?
|
|
}
|
|
|
|
// Hashes returns the supported hash sets.
|
|
func (f *Fs) Hashes() hash.Set {
|
|
return hash.Set(hash.None)
|
|
}
|
|
|
|
// PublicLink generates a public link to the remote path (usually readable by anyone)
|
|
func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration, unlink bool) (link string, err error) {
|
|
root, err := f.findRoot(ctx, false)
|
|
if err != nil {
|
|
return "", errors.Wrap(err, "PublicLink failed to find root node")
|
|
}
|
|
node, err := f.findNode(root, remote)
|
|
if err != nil {
|
|
return "", errors.Wrap(err, "PublicLink failed to find path")
|
|
}
|
|
link, err = f.srv.Link(node, true)
|
|
if err != nil {
|
|
return "", errors.Wrap(err, "PublicLink failed to create link")
|
|
}
|
|
return link, nil
|
|
}
|
|
|
|
// MergeDirs merges the contents of all the directories passed
|
|
// in into the first one and rmdirs the other directories.
|
|
func (f *Fs) MergeDirs(ctx context.Context, dirs []fs.Directory) error {
|
|
if len(dirs) < 2 {
|
|
return nil
|
|
}
|
|
// find dst directory
|
|
dstDir := dirs[0]
|
|
dstDirNode := f.srv.FS.HashLookup(dstDir.ID())
|
|
if dstDirNode == nil {
|
|
return errors.Errorf("MergeDirs failed to find node for: %v", dstDir)
|
|
}
|
|
for _, srcDir := range dirs[1:] {
|
|
// find src directory
|
|
srcDirNode := f.srv.FS.HashLookup(srcDir.ID())
|
|
if srcDirNode == nil {
|
|
return errors.Errorf("MergeDirs failed to find node for: %v", srcDir)
|
|
}
|
|
|
|
// list the objects
|
|
infos := []*mega.Node{}
|
|
_, err := f.list(ctx, srcDirNode, func(info *mega.Node) bool {
|
|
infos = append(infos, info)
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return errors.Wrapf(err, "MergeDirs list failed on %v", srcDir)
|
|
}
|
|
// move them into place
|
|
for _, info := range infos {
|
|
fs.Infof(srcDir, "merging %q", f.opt.Enc.ToStandardName(info.GetName()))
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
err = f.srv.Move(info, dstDirNode)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrapf(err, "MergeDirs move failed on %q in %v", f.opt.Enc.ToStandardName(info.GetName()), srcDir)
|
|
}
|
|
}
|
|
// rmdir (into trash) the now empty source directory
|
|
fs.Infof(srcDir, "removing empty directory")
|
|
err = f.deleteNode(ctx, srcDirNode)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "MergeDirs move failed to rmdir %q", srcDir)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// About gets quota information
|
|
func (f *Fs) About(ctx context.Context) (*fs.Usage, error) {
|
|
var q mega.QuotaResp
|
|
var err error
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
q, err = f.srv.GetQuota()
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "failed to get Mega Quota")
|
|
}
|
|
usage := &fs.Usage{
|
|
Total: fs.NewUsageValue(int64(q.Mstrg)), // quota of bytes that can be used
|
|
Used: fs.NewUsageValue(int64(q.Cstrg)), // bytes in use
|
|
Free: fs.NewUsageValue(int64(q.Mstrg - q.Cstrg)), // bytes which can be uploaded before reaching the quota
|
|
}
|
|
return usage, nil
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// Fs returns the parent Fs
|
|
func (o *Object) Fs() fs.Info {
|
|
return o.fs
|
|
}
|
|
|
|
// Return a string version
|
|
func (o *Object) String() string {
|
|
if o == nil {
|
|
return "<nil>"
|
|
}
|
|
return o.remote
|
|
}
|
|
|
|
// Remote returns the remote path
|
|
func (o *Object) Remote() string {
|
|
return o.remote
|
|
}
|
|
|
|
// Hash returns the hashes of an object
|
|
func (o *Object) Hash(ctx context.Context, t hash.Type) (string, error) {
|
|
return "", hash.ErrUnsupported
|
|
}
|
|
|
|
// Size returns the size of an object in bytes
|
|
func (o *Object) Size() int64 {
|
|
return o.info.GetSize()
|
|
}
|
|
|
|
// setMetaData sets the metadata from info
|
|
func (o *Object) setMetaData(info *mega.Node) (err error) {
|
|
if info.GetType() != mega.FILE {
|
|
return fs.ErrorIsDir // all other node types are directories
|
|
}
|
|
o.info = info
|
|
return nil
|
|
}
|
|
|
|
// readMetaData gets the metadata if it hasn't already been fetched
|
|
//
|
|
// it also sets the info
|
|
func (o *Object) readMetaData(ctx context.Context) (err error) {
|
|
if o.info != nil {
|
|
return nil
|
|
}
|
|
info, err := o.fs.readMetaDataForPath(ctx, o.remote)
|
|
if err != nil {
|
|
if err == fs.ErrorDirNotFound {
|
|
err = fs.ErrorObjectNotFound
|
|
}
|
|
return err
|
|
}
|
|
return o.setMetaData(info)
|
|
}
|
|
|
|
// ModTime returns the modification time of the object
|
|
//
|
|
//
|
|
// It attempts to read the objects mtime and if that isn't present the
|
|
// LastModified returned in the http headers
|
|
func (o *Object) ModTime(ctx context.Context) time.Time {
|
|
return o.info.GetTimeStamp()
|
|
}
|
|
|
|
// SetModTime sets the modification time of the local fs object
|
|
func (o *Object) SetModTime(ctx context.Context, modTime time.Time) error {
|
|
return fs.ErrorCantSetModTime
|
|
}
|
|
|
|
// Storable returns a boolean showing whether this object storable
|
|
func (o *Object) Storable() bool {
|
|
return true
|
|
}
|
|
|
|
// openObject represents a download in progress
|
|
type openObject struct {
|
|
ctx context.Context
|
|
mu sync.Mutex
|
|
o *Object
|
|
d *mega.Download
|
|
id int
|
|
skip int64
|
|
chunk []byte
|
|
closed bool
|
|
}
|
|
|
|
// get the next chunk
|
|
func (oo *openObject) getChunk(ctx context.Context) (err error) {
|
|
if oo.id >= oo.d.Chunks() {
|
|
return io.EOF
|
|
}
|
|
var chunk []byte
|
|
err = oo.o.fs.pacer.Call(func() (bool, error) {
|
|
chunk, err = oo.d.DownloadChunk(oo.id)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
oo.id++
|
|
oo.chunk = chunk
|
|
return nil
|
|
}
|
|
|
|
// Read reads up to len(p) bytes into p.
|
|
func (oo *openObject) Read(p []byte) (n int, err error) {
|
|
oo.mu.Lock()
|
|
defer oo.mu.Unlock()
|
|
if oo.closed {
|
|
return 0, errors.New("read on closed file")
|
|
}
|
|
// Skip data at the start if requested
|
|
for oo.skip > 0 {
|
|
_, size, err := oo.d.ChunkLocation(oo.id)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if oo.skip < int64(size) {
|
|
break
|
|
}
|
|
oo.id++
|
|
oo.skip -= int64(size)
|
|
}
|
|
if len(oo.chunk) == 0 {
|
|
err = oo.getChunk(oo.ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if oo.skip > 0 {
|
|
oo.chunk = oo.chunk[oo.skip:]
|
|
oo.skip = 0
|
|
}
|
|
}
|
|
n = copy(p, oo.chunk)
|
|
oo.chunk = oo.chunk[n:]
|
|
return n, nil
|
|
}
|
|
|
|
// Close closed the file - MAC errors are reported here
|
|
func (oo *openObject) Close() (err error) {
|
|
oo.mu.Lock()
|
|
defer oo.mu.Unlock()
|
|
if oo.closed {
|
|
return nil
|
|
}
|
|
err = oo.o.fs.pacer.Call(func() (bool, error) {
|
|
err = oo.d.Finish()
|
|
return shouldRetry(oo.ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "failed to finish download")
|
|
}
|
|
oo.closed = true
|
|
return nil
|
|
}
|
|
|
|
// Open an object for read
|
|
func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.ReadCloser, err error) {
|
|
var offset, limit int64 = 0, -1
|
|
for _, option := range options {
|
|
switch x := option.(type) {
|
|
case *fs.SeekOption:
|
|
offset = x.Offset
|
|
case *fs.RangeOption:
|
|
offset, limit = x.Decode(o.Size())
|
|
default:
|
|
if option.Mandatory() {
|
|
fs.Logf(o, "Unsupported mandatory option: %v", option)
|
|
}
|
|
}
|
|
}
|
|
|
|
var d *mega.Download
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
d, err = o.fs.srv.NewDownload(o.info)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "open download file failed")
|
|
}
|
|
|
|
oo := &openObject{
|
|
ctx: ctx,
|
|
o: o,
|
|
d: d,
|
|
skip: offset,
|
|
}
|
|
|
|
return readers.NewLimitedReadCloser(oo, limit), nil
|
|
}
|
|
|
|
// Update the object with the contents of the io.Reader, modTime and size
|
|
//
|
|
// If existing is set then it updates the object rather than creating a new one
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (err error) {
|
|
size := src.Size()
|
|
if size < 0 {
|
|
return errors.New("mega backend can't upload a file of unknown length")
|
|
}
|
|
//modTime := src.ModTime(ctx)
|
|
remote := o.Remote()
|
|
|
|
// Create the parent directory
|
|
dirNode, leaf, err := o.fs.mkdirParent(ctx, remote)
|
|
if err != nil {
|
|
return errors.Wrap(err, "update make parent dir failed")
|
|
}
|
|
|
|
var u *mega.Upload
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
u, err = o.fs.srv.NewUpload(dirNode, o.fs.opt.Enc.FromStandardName(leaf), size)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "upload file failed to create session")
|
|
}
|
|
|
|
// Upload the chunks
|
|
// FIXME do this in parallel
|
|
for id := 0; id < u.Chunks(); id++ {
|
|
_, chunkSize, err := u.ChunkLocation(id)
|
|
if err != nil {
|
|
return errors.Wrap(err, "upload failed to read chunk location")
|
|
}
|
|
chunk := make([]byte, chunkSize)
|
|
_, err = io.ReadFull(in, chunk)
|
|
if err != nil {
|
|
return errors.Wrap(err, "upload failed to read data")
|
|
}
|
|
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
err = u.UploadChunk(id, chunk)
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "upload file failed to upload chunk")
|
|
}
|
|
}
|
|
|
|
// Finish the upload
|
|
var info *mega.Node
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
info, err = u.Finish()
|
|
return shouldRetry(ctx, err)
|
|
})
|
|
if err != nil {
|
|
return errors.Wrap(err, "failed to finish upload")
|
|
}
|
|
|
|
// If the upload succeeded and the original object existed, then delete it
|
|
if o.info != nil {
|
|
err = o.fs.deleteNode(ctx, o.info)
|
|
if err != nil {
|
|
return errors.Wrap(err, "upload failed to remove old version")
|
|
}
|
|
o.info = nil
|
|
}
|
|
|
|
return o.setMetaData(info)
|
|
}
|
|
|
|
// Remove an object
|
|
func (o *Object) Remove(ctx context.Context) error {
|
|
err := o.fs.deleteNode(ctx, o.info)
|
|
if err != nil {
|
|
return errors.Wrap(err, "Remove object failed")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ID returns the ID of the Object if known, or "" if not
|
|
func (o *Object) ID() string {
|
|
return o.info.GetHash()
|
|
}
|
|
|
|
// Check the interfaces are satisfied
|
|
var (
|
|
_ fs.Fs = (*Fs)(nil)
|
|
_ fs.Purger = (*Fs)(nil)
|
|
_ fs.Mover = (*Fs)(nil)
|
|
_ fs.PutUncheckeder = (*Fs)(nil)
|
|
_ fs.DirMover = (*Fs)(nil)
|
|
_ fs.DirCacheFlusher = (*Fs)(nil)
|
|
_ fs.PublicLinker = (*Fs)(nil)
|
|
_ fs.MergeDirser = (*Fs)(nil)
|
|
_ fs.Abouter = (*Fs)(nil)
|
|
_ fs.Object = (*Object)(nil)
|
|
_ fs.IDer = (*Object)(nil)
|
|
)
|