[Move Object] Bucket handle integration (#2840)

* move object api integration

* small fix

* review comment

* formating

* review comment

* review comment

* unit test

* small fix

* add one condition

* add one more assert

* formating

* formating

* formating

internal/storage/bucket_handle.go

@@ -549,9 +549,58 @@ func (bh *bucketHandle) DeleteFolder(ctx context.Context, folderName string) (er
 	return err
 }
 
+func isPreconditionFailed(err error) (bool, error) {
+	var gapiErr *googleapi.Error
+	if errors.As(err, &gapiErr) && gapiErr.Code == http.StatusPreconditionFailed {
+		return true, &gcs.PreconditionError{Err: gapiErr}
+	}
+
+	var apiErr *apierror.APIError
+	if errors.As(err, &apiErr) && apiErr.GRPCStatus().Code() == codes.FailedPrecondition {
+		return true, &gcs.PreconditionError{Err: apiErr}
+	}
+
+	return false, nil
+}
+
 func (bh *bucketHandle) MoveObject(ctx context.Context, req *gcs.MoveObjectRequest) (*gcs.Object, error) {
-	// TODO: Implement it.
-	return nil, nil
+	var o *gcs.Object
+	var err error
+
+	obj := bh.bucket.Object(req.SrcName)
+
+	// Switching to the requested generation of source object.
+	if req.SrcGeneration != 0 {
+		obj = obj.Generation(req.SrcGeneration)
+	}
+
+	// Putting a condition that the metaGeneration of source should match *req.SrcMetaGenerationPrecondition for move operation to occur.
+	if req.SrcMetaGenerationPrecondition != nil {
+		obj = obj.If(storage.Conditions{MetagenerationMatch: *req.SrcMetaGenerationPrecondition})
+	}
+
+	dstMoveObject := storage.MoveObjectDestination{
+		Object:     req.DstName,
+		Conditions: nil,
+	}
+
+	attrs, err := obj.Move(ctx, dstMoveObject)
+	if err == nil {
+		// Converting objAttrs to type *Object
+		o = storageutil.ObjectAttrsToBucketObject(attrs)
+		return o, nil
+	}
+
+	// If storage object does not exist, httpclient is returning ErrObjectNotExist error instead of googleapi error
+	// https://github.com/GoogleCloudPlatform/gcsfuse/blob/master/vendor/cloud.google.com/go/storage/http_client.go#L516
+	if ok, preCondErr := isPreconditionFailed(err); ok {
+		err = preCondErr
+	} else if errors.Is(err, storage.ErrObjectNotExist) {
+		err = &gcs.NotFoundError{Err: storage.ErrObjectNotExist}
+	} else {
+		err = fmt.Errorf("error in moving object: %w", err)
+	}
+	return nil, err
 }
 
 func (bh *bucketHandle) RenameFolder(ctx context.Context, folderName string, destinationFolderId string) (folder *gcs.Folder, err error) {

internal/storage/bucket_handle_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"reflect"
 	"strings"
 	"testing"
@@ -26,11 +27,13 @@ import (
 	"cloud.google.com/go/storage"
 	control "cloud.google.com/go/storage/control/apiv2"
 	"cloud.google.com/go/storage/control/apiv2/controlpb"
+	"github.com/googleapis/gax-go/v2/apierror"
 	"github.com/googlecloudplatform/gcsfuse/v2/internal/storage/gcs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -1520,3 +1523,59 @@ func (testSuite *BucketHandleTest) TestCreateFolderWithGivenName() {
 	assert.NoError(testSuite.T(), err)
 	assert.Equal(testSuite.T(), gcs.GCSFolder(TestBucketName, &mockFolder), folder)
 }
+
+func TestIsPreconditionFailed(t *testing.T) {
+	preCondApiError, _ := apierror.FromError(status.New(codes.FailedPrecondition, "Precondition error").Err())
+	notFoundApiError, _ := apierror.FromError(status.New(codes.NotFound, "Not Found error").Err())
+
+	tests := []struct {
+		name          string
+		err           error
+		expectPreCond bool
+	}{
+		{
+			name:          "googleapi.Error with PreconditionFailed",
+			err:           &googleapi.Error{Code: http.StatusPreconditionFailed},
+			expectPreCond: true,
+		},
+		{
+			name:          "googleapi.Error with other code",
+			err:           &googleapi.Error{Code: http.StatusNotFound},
+			expectPreCond: false,
+		},
+		{
+			name:          "apierror.APIError with FailedPrecondition",
+			err:           preCondApiError,
+			expectPreCond: true,
+		},
+		{
+			name:          "apierror.APIError with other code",
+			err:           notFoundApiError,
+			expectPreCond: false,
+		},
+		{
+			name:          "nil error",
+			err:           nil,
+			expectPreCond: false,
+		},
+		{
+			name:          "generic error",
+			err:           errors.New("generic error"),
+			expectPreCond: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			isPreCond, err := isPreconditionFailed(tt.err)
+
+			assert.Equal(t, tt.expectPreCond, isPreCond)
+			if tt.expectPreCond {
+				var preCondErr *gcs.PreconditionError
+				assert.ErrorAs(t, err, &preCondErr)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}