sftpgo/internal/vfs/gcsfs.go

// Copyright (C) 2019 Nicola Murino
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, version 3.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//go:build !nogcs
// +build !nogcs

package vfs

import (
	"context"
	"fmt"
	"io"
	"mime"
	"net/http"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"cloud.google.com/go/storage"
	"github.com/eikenb/pipeat"
	"github.com/pkg/sftp"
	"github.com/rs/xid"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"

	"github.com/drakkan/sftpgo/v2/internal/logger"
	"github.com/drakkan/sftpgo/v2/internal/metric"
	"github.com/drakkan/sftpgo/v2/internal/util"
	"github.com/drakkan/sftpgo/v2/internal/version"
)

const (
	defaultGCSPageSize = 5000
)

var (
	gcsDefaultFieldsSelection = []string{"Name", "Size", "Deleted", "Updated", "ContentType", "Metadata"}
)

// GCSFs is a Fs implementation for Google Cloud Storage.
type GCSFs struct {
	connectionID string
	localTempDir string
	// if not empty this fs is mouted as virtual folder in the specified path
	mountPath      string
	config         *GCSFsConfig
	svc            *storage.Client
	ctxTimeout     time.Duration
	ctxLongTimeout time.Duration
}

func init() {
	version.AddFeature("+gcs")
}

// NewGCSFs returns an GCSFs object that allows to interact with Google Cloud Storage
func NewGCSFs(connectionID, localTempDir, mountPath string, config GCSFsConfig) (Fs, error) {
	if localTempDir == "" {
		localTempDir = getLocalTempDir()
	}

	var err error
	fs := &GCSFs{
		connectionID:   connectionID,
		localTempDir:   localTempDir,
		mountPath:      getMountPath(mountPath),
		config:         &config,
		ctxTimeout:     30 * time.Second,
		ctxLongTimeout: 300 * time.Second,
	}
	if err = fs.config.validate(); err != nil {
		return fs, err
	}
	ctx := context.Background()
	if fs.config.AutomaticCredentials > 0 {
		fs.svc, err = storage.NewClient(ctx)
	} else {
		err = fs.config.Credentials.TryDecrypt()
		if err != nil {
			return fs, err
		}
		fs.svc, err = storage.NewClient(ctx, option.WithCredentialsJSON([]byte(fs.config.Credentials.GetPayload())))
	}
	return fs, err
}

// Name returns the name for the Fs implementation
func (fs *GCSFs) Name() string {
	return fmt.Sprintf("%s bucket %q", gcsfsName, fs.config.Bucket)
}

// ConnectionID returns the connection ID associated to this Fs implementation
func (fs *GCSFs) ConnectionID() string {
	return fs.connectionID
}

// Stat returns a FileInfo describing the named file
func (fs *GCSFs) Stat(name string) (os.FileInfo, error) {
	if name == "" || name == "/" || name == "." {
		return NewFileInfo(name, true, 0, time.Unix(0, 0), false), nil
	}
	if fs.config.KeyPrefix == name+"/" {
		return NewFileInfo(name, true, 0, time.Unix(0, 0), false), nil
	}
	return fs.getObjectStat(name)
}

// Lstat returns a FileInfo describing the named file
func (fs *GCSFs) Lstat(name string) (os.FileInfo, error) {
	return fs.Stat(name)
}

// Open opens the named file for reading
func (fs *GCSFs) Open(name string, offset int64) (File, PipeReader, func(), error) {
	r, w, err := pipeat.PipeInDir(fs.localTempDir)
	if err != nil {
		return nil, nil, nil, err
	}
	p := NewPipeReader(r)
	if readMetadata > 0 {
		attrs, err := fs.headObject(name)
		if err != nil {
			r.Close()
			w.Close()
			return nil, nil, nil, err
		}
		p.setMetadata(attrs.Metadata)
	}
	bkt := fs.svc.Bucket(fs.config.Bucket)
	obj := bkt.Object(name)
	ctx, cancelFn := context.WithCancel(context.Background())
	objectReader, err := obj.NewRangeReader(ctx, offset, -1)
	if err == nil && offset > 0 && objectReader.Attrs.ContentEncoding == "gzip" {
		err = fmt.Errorf("range request is not possible for gzip content encoding, requested offset %d", offset)
		objectReader.Close()
	}
	if err != nil {
		r.Close()
		w.Close()
		cancelFn()
		return nil, nil, nil, err
	}
	go func() {
		defer cancelFn()
		defer objectReader.Close()

		n, err := io.Copy(w, objectReader)
		w.CloseWithError(err) //nolint:errcheck
		fsLog(fs, logger.LevelDebug, "download completed, path: %q size: %v, err: %+v", name, n, err)
		metric.GCSTransferCompleted(n, 1, err)
	}()
	return nil, p, cancelFn, nil
}

// Create creates or opens the named file for writing
func (fs *GCSFs) Create(name string, flag, checks int) (File, PipeWriter, func(), error) {
	if checks&CheckParentDir != 0 {
		_, err := fs.Stat(path.Dir(name))
		if err != nil {
			return nil, nil, nil, err
		}
	}
	r, w, err := pipeat.PipeInDir(fs.localTempDir)
	if err != nil {
		return nil, nil, nil, err
	}
	var partialFileName string
	var attrs *storage.ObjectAttrs
	var statErr error

	bkt := fs.svc.Bucket(fs.config.Bucket)
	obj := bkt.Object(name)

	if flag == -1 {
		obj = obj.If(storage.Conditions{DoesNotExist: true})
	} else {
		attrs, statErr = fs.headObject(name)
		if statErr == nil {
			obj = obj.If(storage.Conditions{GenerationMatch: attrs.Generation})
		} else if fs.IsNotExist(statErr) {
			obj = obj.If(storage.Conditions{DoesNotExist: true})
		} else {
			fsLog(fs, logger.LevelWarn, "unable to set precondition for %q, stat err: %v", name, statErr)
		}
	}
	ctx, cancelFn := context.WithCancel(context.Background())

	var p PipeWriter
	var objectWriter *storage.Writer
	if checks&CheckResume != 0 {
		if statErr != nil {
			cancelFn()
			r.Close()
			w.Close()
			return nil, nil, nil, fmt.Errorf("unable to resume %q stat error: %w", name, statErr)
		}
		p = newPipeWriterAtOffset(w, attrs.Size)
		partialFileName = fs.getTempObject(name)
		partialObj := bkt.Object(partialFileName)
		partialObj = partialObj.If(storage.Conditions{DoesNotExist: true})
		objectWriter = partialObj.NewWriter(ctx)
	} else {
		p = NewPipeWriter(w)
		objectWriter = obj.NewWriter(ctx)
	}

	if fs.config.UploadPartSize > 0 {
		objectWriter.ChunkSize = int(fs.config.UploadPartSize) * 1024 * 1024
	}
	if fs.config.UploadPartMaxTime > 0 {
		objectWriter.ChunkRetryDeadline = time.Duration(fs.config.UploadPartMaxTime) * time.Second
	}
	fs.setWriterAttrs(objectWriter, flag, name)

	go func() {
		defer cancelFn()

		n, err := io.Copy(objectWriter, r)
		closeErr := objectWriter.Close()
		if err == nil {
			err = closeErr
		}
		if err == nil && partialFileName != "" {
			partialObject := bkt.Object(partialFileName)
			partialObject = partialObject.If(storage.Conditions{GenerationMatch: objectWriter.Attrs().Generation})
			err = fs.composeObjects(ctx, obj, partialObject)
		}
		r.CloseWithError(err) //nolint:errcheck
		p.Done(err)
		fsLog(fs, logger.LevelDebug, "upload completed, path: %q, acl: %q, readed bytes: %v, err: %+v",
			name, fs.config.ACL, n, err)
		metric.GCSTransferCompleted(n, 0, err)
	}()

	if uploadMode&8 != 0 {
		return nil, p, nil, nil
	}
	return nil, p, cancelFn, nil
}

// Rename renames (moves) source to target.
func (fs *GCSFs) Rename(source, target string) (int, int64, error) {
	if source == target {
		return -1, -1, nil
	}
	_, err := fs.Stat(path.Dir(target))
	if err != nil {
		return -1, -1, err
	}
	fi, err := fs.getObjectStat(source)
	if err != nil {
		return -1, -1, err
	}
	return fs.renameInternal(source, target, fi, 0)
}

// Remove removes the named file or (empty) directory.
func (fs *GCSFs) Remove(name string, isDir bool) error {
	if isDir {
		hasContents, err := fs.hasContents(name)
		if err != nil {
			return err
		}
		if hasContents {
			return fmt.Errorf("cannot remove non empty directory: %q", name)
		}
		if !strings.HasSuffix(name, "/") {
			name += "/"
		}
	}
	obj := fs.svc.Bucket(fs.config.Bucket).Object(name)
	attrs, statErr := fs.headObject(name)
	if statErr == nil {
		obj = obj.If(storage.Conditions{GenerationMatch: attrs.Generation})
	} else {
		fsLog(fs, logger.LevelWarn, "unable to set precondition for deleting %q, stat err: %v",
			name, statErr)
	}

	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
	defer cancelFn()

	err := obj.Delete(ctx)
	if isDir && fs.IsNotExist(err) {
		// we can have directories without a trailing "/" (created using v2.1.0 and before)
		ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
		defer cancelFn()

		err = fs.svc.Bucket(fs.config.Bucket).Object(strings.TrimSuffix(name, "/")).Delete(ctx)
	}
	metric.GCSDeleteObjectCompleted(err)
	return err
}

// Mkdir creates a new directory with the specified name and default permissions
func (fs *GCSFs) Mkdir(name string) error {
	_, err := fs.Stat(name)
	if !fs.IsNotExist(err) {
		return err
	}
	return fs.mkdirInternal(name)
}

// Symlink creates source as a symbolic link to target.
func (*GCSFs) Symlink(_, _ string) error {
	return ErrVfsUnsupported
}

// Readlink returns the destination of the named symbolic link
func (*GCSFs) Readlink(_ string) (string, error) {
	return "", ErrVfsUnsupported
}

// Chown changes the numeric uid and gid of the named file.
func (*GCSFs) Chown(_ string, _ int, _ int) error {
	return ErrVfsUnsupported
}

// Chmod changes the mode of the named file to mode.
func (*GCSFs) Chmod(_ string, _ os.FileMode) error {
	return ErrVfsUnsupported
}

// Chtimes changes the access and modification times of the named file.
func (fs *GCSFs) Chtimes(name string, _, mtime time.Time, isUploading bool) error {
	if isUploading {
		return nil
	}
	obj := fs.svc.Bucket(fs.config.Bucket).Object(name)
	attrs, err := fs.headObject(name)
	if err != nil {
		return err
	}
	obj = obj.If(storage.Conditions{MetagenerationMatch: attrs.Metageneration})

	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
	defer cancelFn()

	metadata := attrs.Metadata
	if metadata == nil {
		metadata = make(map[string]string)
	}
	metadata[lastModifiedField] = strconv.FormatInt(mtime.UnixMilli(), 10)

	objectAttrsToUpdate := storage.ObjectAttrsToUpdate{
		Metadata: metadata,
	}
	_, err = obj.Update(ctx, objectAttrsToUpdate)

	return err
}

// Truncate changes the size of the named file.
// Truncate by path is not supported, while truncating an opened
// file is handled inside base transfer
func (*GCSFs) Truncate(_ string, _ int64) error {
	return ErrVfsUnsupported
}

// ReadDir reads the directory named by dirname and returns
// a list of directory entries.
func (fs *GCSFs) ReadDir(dirname string) (DirLister, error) {
	// dirname must be already cleaned
	prefix := fs.getPrefix(dirname)
	query := &storage.Query{Prefix: prefix, Delimiter: "/"}
	err := query.SetAttrSelection(gcsDefaultFieldsSelection)
	if err != nil {
		return nil, err
	}
	bkt := fs.svc.Bucket(fs.config.Bucket)

	return &gcsDirLister{
		bucket:   bkt,
		query:    query,
		timeout:  fs.ctxTimeout,
		prefix:   prefix,
		prefixes: make(map[string]bool),
	}, nil
}

// IsUploadResumeSupported returns true if resuming uploads is supported.
// Resuming uploads is not supported on GCS
func (*GCSFs) IsUploadResumeSupported() bool {
	return false
}

// IsConditionalUploadResumeSupported returns if resuming uploads is supported
// for the specified size
func (*GCSFs) IsConditionalUploadResumeSupported(_ int64) bool {
	return true
}

// IsAtomicUploadSupported returns true if atomic upload is supported.
// S3 uploads are already atomic, we don't need to upload to a temporary
// file
func (*GCSFs) IsAtomicUploadSupported() bool {
	return false
}

// IsNotExist returns a boolean indicating whether the error is known to
// report that a file or directory does not exist
func (*GCSFs) IsNotExist(err error) bool {
	if err == nil {
		return false
	}
	if err == storage.ErrObjectNotExist || err == storage.ErrBucketNotExist {
		return true
	}
	if e, ok := err.(*googleapi.Error); ok {
		if e.Code == http.StatusNotFound {
			return true
		}
	}
	return false
}

// IsPermission returns a boolean indicating whether the error is known to
// report that permission is denied.
func (*GCSFs) IsPermission(err error) bool {
	if err == nil {
		return false
	}
	if e, ok := err.(*googleapi.Error); ok {
		if e.Code == http.StatusForbidden || e.Code == http.StatusUnauthorized {
			return true
		}
	}
	return false
}

// IsNotSupported returns true if the error indicate an unsupported operation
func (*GCSFs) IsNotSupported(err error) bool {
	if err == nil {
		return false
	}
	return err == ErrVfsUnsupported
}

// CheckRootPath creates the specified local root directory if it does not exists
func (fs *GCSFs) CheckRootPath(username string, uid int, gid int) bool {
	// we need a local directory for temporary files
	osFs := NewOsFs(fs.ConnectionID(), fs.localTempDir, "", nil)
	return osFs.CheckRootPath(username, uid, gid)
}

// ScanRootDirContents returns the number of files contained in the bucket,
// and their size
func (fs *GCSFs) ScanRootDirContents() (int, int64, error) {
	return fs.GetDirSize(fs.config.KeyPrefix)
}

// GetDirSize returns the number of files and the size for a folder
// including any subfolders
func (fs *GCSFs) GetDirSize(dirname string) (int, int64, error) {
	prefix := fs.getPrefix(dirname)
	numFiles := 0
	size := int64(0)

	query := &storage.Query{Prefix: prefix}
	err := query.SetAttrSelection(gcsDefaultFieldsSelection)
	if err != nil {
		return numFiles, size, err
	}

	iteratePage := func(nextPageToken string) (string, error) {
		ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
		defer cancelFn()

		bkt := fs.svc.Bucket(fs.config.Bucket)
		it := bkt.Objects(ctx, query)
		pager := iterator.NewPager(it, defaultGCSPageSize, nextPageToken)

		var objects []*storage.ObjectAttrs
		pageToken, err := pager.NextPage(&objects)
		if err != nil {
			return pageToken, err
		}
		for _, attrs := range objects {
			if !attrs.Deleted.IsZero() {
				continue
			}
			isDir := strings.HasSuffix(attrs.Name, "/") || attrs.ContentType == dirMimeType
			if isDir && attrs.Size == 0 {
				continue
			}
			numFiles++
			size += attrs.Size
		}
		return pageToken, nil
	}

	pageToken := ""
	for {
		pageToken, err = iteratePage(pageToken)
		if err != nil {
			metric.GCSListObjectsCompleted(err)
			return numFiles, size, err
		}
		fsLog(fs, logger.LevelDebug, "scan in progress for %q, files: %d, size: %d", dirname, numFiles, size)
		if pageToken == "" {
			break
		}
	}

	metric.GCSListObjectsCompleted(nil)
	return numFiles, size, err
}

// GetAtomicUploadPath returns the path to use for an atomic upload.
// GCS uploads are already atomic, we never call this method for GCS
func (*GCSFs) GetAtomicUploadPath(_ string) string {
	return ""
}

// GetRelativePath returns the path for a file relative to the user's home dir.
// This is the path as seen by SFTPGo users
func (fs *GCSFs) GetRelativePath(name string) string {
	rel := path.Clean(name)
	if rel == "." {
		rel = ""
	}
	if !path.IsAbs(rel) {
		rel = "/" + rel
	}
	if fs.config.KeyPrefix != "" {
		if !strings.HasPrefix(rel, "/"+fs.config.KeyPrefix) {
			rel = "/"
		}
		rel = path.Clean("/" + strings.TrimPrefix(rel, "/"+fs.config.KeyPrefix))
	}
	if fs.mountPath != "" {
		rel = path.Join(fs.mountPath, rel)
	}
	return rel
}

// Walk walks the file tree rooted at root, calling walkFn for each file or
// directory in the tree, including root
func (fs *GCSFs) Walk(root string, walkFn filepath.WalkFunc) error {
	prefix := fs.getPrefix(root)

	query := &storage.Query{Prefix: prefix}
	err := query.SetAttrSelection(gcsDefaultFieldsSelection)
	if err != nil {
		walkFn(root, nil, err) //nolint:errcheck
		return err
	}

	iteratePage := func(nextPageToken string) (string, error) {
		ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
		defer cancelFn()

		bkt := fs.svc.Bucket(fs.config.Bucket)
		it := bkt.Objects(ctx, query)
		pager := iterator.NewPager(it, defaultGCSPageSize, nextPageToken)

		var objects []*storage.ObjectAttrs
		pageToken, err := pager.NextPage(&objects)
		if err != nil {
			walkFn(root, nil, err) //nolint:errcheck
			return pageToken, err
		}
		for _, attrs := range objects {
			if !attrs.Deleted.IsZero() {
				continue
			}
			name, isDir := fs.resolve(attrs.Name, prefix, attrs.ContentType)
			if name == "" {
				continue
			}
			objectModTime := attrs.Updated
			if val := getLastModified(attrs.Metadata); val > 0 {
				objectModTime = util.GetTimeFromMsecSinceEpoch(val)
			}
			err = walkFn(attrs.Name, NewFileInfo(name, isDir, attrs.Size, objectModTime, false), nil)
			if err != nil {
				return pageToken, err
			}
		}

		return pageToken, nil
	}

	pageToken := ""
	for {
		pageToken, err = iteratePage(pageToken)
		if err != nil {
			metric.GCSListObjectsCompleted(err)
			return err
		}
		if pageToken == "" {
			break
		}
	}

	walkFn(root, NewFileInfo(root, true, 0, time.Unix(0, 0), false), err) //nolint:errcheck
	metric.GCSListObjectsCompleted(err)
	return err
}

// Join joins any number of path elements into a single path
func (*GCSFs) Join(elem ...string) string {
	return strings.TrimPrefix(path.Join(elem...), "/")
}

// HasVirtualFolders returns true if folders are emulated
func (GCSFs) HasVirtualFolders() bool {
	return true
}

// ResolvePath returns the matching filesystem path for the specified virtual path
func (fs *GCSFs) ResolvePath(virtualPath string) (string, error) {
	if fs.mountPath != "" {
		virtualPath = strings.TrimPrefix(virtualPath, fs.mountPath)
	}
	if !path.IsAbs(virtualPath) {
		virtualPath = path.Clean("/" + virtualPath)
	}
	return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(virtualPath, "/")), nil
}

// CopyFile implements the FsFileCopier interface
func (fs *GCSFs) CopyFile(source, target string, _ int64) error {
	return fs.copyFileInternal(source, target)
}

func (fs *GCSFs) resolve(name, prefix, contentType string) (string, bool) {
	result := strings.TrimPrefix(name, prefix)
	isDir := strings.HasSuffix(result, "/")
	if isDir {
		result = strings.TrimSuffix(result, "/")
	}
	if contentType == dirMimeType {
		isDir = true
	}
	return result, isDir
}

// getObjectStat returns the stat result
func (fs *GCSFs) getObjectStat(name string) (os.FileInfo, error) {
	attrs, err := fs.headObject(name)
	if err == nil {
		objSize := attrs.Size
		objectModTime := attrs.Updated
		if val := getLastModified(attrs.Metadata); val > 0 {
			objectModTime = util.GetTimeFromMsecSinceEpoch(val)
		}
		isDir := attrs.ContentType == dirMimeType || strings.HasSuffix(attrs.Name, "/")
		return NewFileInfo(name, isDir, objSize, objectModTime, false), nil
	}
	if !fs.IsNotExist(err) {
		return nil, err
	}
	// now check if this is a prefix (virtual directory)
	hasContents, err := fs.hasContents(name)
	if err != nil {
		return nil, err
	}
	if hasContents {
		return NewFileInfo(name, true, 0, time.Unix(0, 0), false), nil
	}
	// finally check if this is an object with a trailing /
	attrs, err = fs.headObject(name + "/")
	if err != nil {
		return nil, err
	}
	objectModTime := attrs.Updated
	if val := getLastModified(attrs.Metadata); val > 0 {
		objectModTime = util.GetTimeFromMsecSinceEpoch(val)
	}
	return NewFileInfo(name, true, attrs.Size, objectModTime, false), nil
}

func (fs *GCSFs) setWriterAttrs(objectWriter *storage.Writer, flag int, name string) {
	var contentType string
	if flag == -1 {
		contentType = dirMimeType
	} else {
		contentType = mime.TypeByExtension(path.Ext(name))
	}
	if contentType != "" {
		objectWriter.ObjectAttrs.ContentType = contentType
	}
	if fs.config.StorageClass != "" {
		objectWriter.ObjectAttrs.StorageClass = fs.config.StorageClass
	}
	if fs.config.ACL != "" {
		objectWriter.PredefinedACL = fs.config.ACL
	}
}

func (fs *GCSFs) composeObjects(ctx context.Context, dst, partialObject *storage.ObjectHandle) error {
	fsLog(fs, logger.LevelDebug, "start object compose for partial file %q, destination %q",
		partialObject.ObjectName(), dst.ObjectName())
	composer := dst.ComposerFrom(dst, partialObject)
	if fs.config.StorageClass != "" {
		composer.StorageClass = fs.config.StorageClass
	}
	if fs.config.ACL != "" {
		composer.PredefinedACL = fs.config.ACL
	}
	contentType := mime.TypeByExtension(path.Ext(dst.ObjectName()))
	if contentType != "" {
		composer.ContentType = contentType
	}
	_, err := composer.Run(ctx)
	fsLog(fs, logger.LevelDebug, "object compose for %q finished, err: %v", dst.ObjectName(), err)

	delCtx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
	defer cancelFn()

	errDelete := partialObject.Delete(delCtx)
	metric.GCSDeleteObjectCompleted(errDelete)
	fsLog(fs, logger.LevelDebug, "deleted partial file %q after composing with %q, err: %v",
		partialObject.ObjectName(), dst.ObjectName(), errDelete)
	return err
}

func (fs *GCSFs) copyFileInternal(source, target string) error {
	src := fs.svc.Bucket(fs.config.Bucket).Object(source)
	dst := fs.svc.Bucket(fs.config.Bucket).Object(target)
	attrs, statErr := fs.headObject(target)
	if statErr == nil {
		dst = dst.If(storage.Conditions{GenerationMatch: attrs.Generation})
	} else if fs.IsNotExist(statErr) {
		dst = dst.If(storage.Conditions{DoesNotExist: true})
	} else {
		fsLog(fs, logger.LevelWarn, "unable to set precondition for rename, target %q, stat err: %v",
			target, statErr)
	}

	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxLongTimeout))
	defer cancelFn()

	copier := dst.CopierFrom(src)
	if fs.config.StorageClass != "" {
		copier.StorageClass = fs.config.StorageClass
	}
	if fs.config.ACL != "" {
		copier.PredefinedACL = fs.config.ACL
	}
	contentType := mime.TypeByExtension(path.Ext(source))
	if contentType != "" {
		copier.ContentType = contentType
	}
	_, err := copier.Run(ctx)
	metric.GCSCopyObjectCompleted(err)
	return err
}

func (fs *GCSFs) renameInternal(source, target string, fi os.FileInfo, recursion int) (int, int64, error) {
	var numFiles int
	var filesSize int64

	if fi.IsDir() {
		if renameMode == 0 {
			hasContents, err := fs.hasContents(source)
			if err != nil {
				return numFiles, filesSize, err
			}
			if hasContents {
				return numFiles, filesSize, fmt.Errorf("cannot rename non empty directory: %q", source)
			}
		}
		if err := fs.mkdirInternal(target); err != nil {
			return numFiles, filesSize, err
		}
		if renameMode == 1 {
			files, size, err := doRecursiveRename(fs, source, target, fs.renameInternal, recursion)
			numFiles += files
			filesSize += size
			if err != nil {
				return numFiles, filesSize, err
			}
		}
	} else {
		if err := fs.copyFileInternal(source, target); err != nil {
			return numFiles, filesSize, err
		}
		numFiles++
		filesSize += fi.Size()
	}
	err := fs.Remove(source, fi.IsDir())
	if fs.IsNotExist(err) {
		err = nil
	}
	return numFiles, filesSize, err
}

func (fs *GCSFs) mkdirInternal(name string) error {
	if !strings.HasSuffix(name, "/") {
		name += "/"
	}
	_, w, _, err := fs.Create(name, -1, 0)
	if err != nil {
		return err
	}
	return w.Close()
}

func (fs *GCSFs) hasContents(name string) (bool, error) {
	result := false
	prefix := fs.getPrefix(name)
	query := &storage.Query{Prefix: prefix}
	err := query.SetAttrSelection(gcsDefaultFieldsSelection)
	if err != nil {
		return result, err
	}
	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
	defer cancelFn()

	bkt := fs.svc.Bucket(fs.config.Bucket)
	it := bkt.Objects(ctx, query)
	// if we have a dir object with a trailing slash it will be returned so we set the size to 2
	pager := iterator.NewPager(it, 2, "")

	var objects []*storage.ObjectAttrs
	_, err = pager.NextPage(&objects)
	if err != nil {
		metric.GCSListObjectsCompleted(err)
		return result, err
	}

	for _, attrs := range objects {
		name, _ := fs.resolve(attrs.Name, prefix, attrs.ContentType)
		// a dir object with a trailing slash will result in an empty name
		if name == "/" || name == "" {
			continue
		}
		result = true
		break
	}

	metric.GCSListObjectsCompleted(nil)
	return result, nil
}

func (fs *GCSFs) getPrefix(name string) string {
	prefix := ""
	if name != "" && name != "." && name != "/" {
		prefix = strings.TrimPrefix(name, "/")
		if !strings.HasSuffix(prefix, "/") {
			prefix += "/"
		}
	}
	return prefix
}

func (fs *GCSFs) headObject(name string) (*storage.ObjectAttrs, error) {
	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
	defer cancelFn()

	bkt := fs.svc.Bucket(fs.config.Bucket)
	obj := bkt.Object(name)
	attrs, err := obj.Attrs(ctx)
	metric.GCSHeadObjectCompleted(err)
	return attrs, err
}

// GetMimeType returns the content type
func (fs *GCSFs) GetMimeType(name string) (string, error) {
	attrs, err := fs.headObject(name)
	if err != nil {
		return "", err
	}
	return attrs.ContentType, nil
}

// Close closes the fs
func (fs *GCSFs) Close() error {
	return nil
}

// GetAvailableDiskSize returns the available size for the specified path
func (*GCSFs) GetAvailableDiskSize(_ string) (*sftp.StatVFS, error) {
	return nil, ErrStorageSizeUnavailable
}

func (*GCSFs) getTempObject(name string) string {
	dir := filepath.Dir(name)
	guid := xid.New().String()
	return filepath.Join(dir, ".sftpgo-partial."+guid+"."+filepath.Base(name))
}

type gcsDirLister struct {
	baseDirLister
	bucket        *storage.BucketHandle
	query         *storage.Query
	timeout       time.Duration
	nextPageToken string
	noMorePages   bool
	prefix        string
	prefixes      map[string]bool
	metricUpdated bool
}

func (l *gcsDirLister) resolve(name, contentType string) (string, bool) {
	result := strings.TrimPrefix(name, l.prefix)
	isDir := strings.HasSuffix(result, "/")
	if isDir {
		result = strings.TrimSuffix(result, "/")
	}
	if contentType == dirMimeType {
		isDir = true
	}
	return result, isDir
}

func (l *gcsDirLister) Next(limit int) ([]os.FileInfo, error) {
	if limit <= 0 {
		return nil, errInvalidDirListerLimit
	}
	if len(l.cache) >= limit {
		return l.returnFromCache(limit), nil
	}

	if l.noMorePages {
		if !l.metricUpdated {
			l.metricUpdated = true
			metric.GCSListObjectsCompleted(nil)
		}
		return l.returnFromCache(limit), io.EOF
	}

	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(l.timeout))
	defer cancelFn()

	it := l.bucket.Objects(ctx, l.query)
	paginator := iterator.NewPager(it, defaultGCSPageSize, l.nextPageToken)
	var objects []*storage.ObjectAttrs

	pageToken, err := paginator.NextPage(&objects)
	if err != nil {
		metric.GCSListObjectsCompleted(err)
		return l.cache, err
	}

	for _, attrs := range objects {
		if attrs.Prefix != "" {
			name, _ := l.resolve(attrs.Prefix, attrs.ContentType)
			if name == "" {
				continue
			}
			if _, ok := l.prefixes[name]; ok {
				continue
			}
			l.cache = append(l.cache, NewFileInfo(name, true, 0, time.Unix(0, 0), false))
			l.prefixes[name] = true
		} else {
			name, isDir := l.resolve(attrs.Name, attrs.ContentType)
			if name == "" {
				continue
			}
			if !attrs.Deleted.IsZero() {
				continue
			}
			if isDir {
				// check if the dir is already included, it will be sent as blob prefix if it contains at least one item
				if _, ok := l.prefixes[name]; ok {
					continue
				}
				l.prefixes[name] = true
			}
			modTime := attrs.Updated
			if val := getLastModified(attrs.Metadata); val > 0 {
				modTime = util.GetTimeFromMsecSinceEpoch(val)
			}
			l.cache = append(l.cache, NewFileInfo(name, isDir, attrs.Size, modTime, false))
		}
	}

	l.nextPageToken = pageToken
	l.noMorePages = (l.nextPageToken == "")

	return l.returnFromCache(limit), nil
}

func (l *gcsDirLister) Close() error {
	clear(l.prefixes)
	return l.baseDirLister.Close()
}