frostfs-sdk-go/client/object_patch.go

package client

import (
	"context"
	"crypto/ecdsa"
	"errors"
	"fmt"
	"io"

	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/acl"
	v2object "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/object"
	rpcapi "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc/client"
	v2session "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/session"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/signature"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
)

// ObjectPatcher is designed to patch an object.
//
// Must be initialized using Client.ObjectPatchInit, any other
// usage is unsafe.
type ObjectPatcher interface {
	// PatchAttributes patches attributes. Attributes can be patched no more than once,
	// otherwise, the server returns an error.
	//
	// Result means success. Failure reason can be received via Close.
	PatchAttributes(ctx context.Context, newAttrs []object.Attribute, replace bool) bool

	// PatchPayload patches the object's payload.
	//
	// PatchPayload receives `payloadReader` and thus the payload of the patch is read and sent by chunks of
	// `MaxChunkLength` length.
	//
	// Result means success. Failure reason can be received via Close.
	PatchPayload(ctx context.Context, rng *object.Range, payloadReader io.Reader) bool

	// Close ends patching the object and returns the result of the operation
	// along with the final results. Must be called after using the ObjectPatcher.
	//
	// Exactly one return value is non-nil. By default, server status is returned in res structure.
	// Any client's internal or transport errors are returned as Go built-in error.
	// If Client is tuned to resolve FrostFS API statuses, then FrostFS failures
	// codes are returned as error.
	//
	// Return statuses:
	//   - global (see Client docs);
	//   - *apistatus.ContainerNotFound;
	//   - *apistatus.ContainerAccessDenied;
	//   - *apistatus.ObjectAccessDenied;
	//   - *apistatus.ObjectAlreadyRemoved;
	//   - *apistatus.ObjectLocked;
	//   - *apistatus.ObjectOutOfRange;
	//   - *apistatus.SessionTokenNotFound;
	//   - *apistatus.SessionTokenExpired.
	Close(_ context.Context) (*ResObjectPatch, error)
}

// ResObjectPatch groups resulting values of ObjectPatch operation.
type ResObjectPatch struct {
	statusRes

	obj oid.ID
}

// ObjectID returns an object ID of the patched object.
func (r ResObjectPatch) ObjectID() oid.ID {
	return r.obj
}

// PrmObjectPatch groups parameters of ObjectPatch operation.
type PrmObjectPatch struct {
	XHeaders []string

	Address oid.Address

	BearerToken *bearer.Token

	Session *session.Object

	Key *ecdsa.PrivateKey

	MaxChunkLength int
}

// ObjectPatchInit initializes object patcher.
func (c *Client) ObjectPatchInit(ctx context.Context, prm PrmObjectPatch) (ObjectPatcher, error) {
	if len(prm.XHeaders)%2 != 0 {
		return nil, errorInvalidXHeaders
	}

	var objectPatcher objectPatcher
	stream, err := rpcapi.Patch(&c.c, &objectPatcher.respV2, client.WithContext(ctx))
	if err != nil {
		return nil, fmt.Errorf("open stream: %w", err)
	}

	objectPatcher.addr = prm.Address
	objectPatcher.key = &c.prm.Key
	if prm.Key != nil {
		objectPatcher.key = prm.Key
	}
	objectPatcher.client = c
	objectPatcher.stream = stream

	if prm.MaxChunkLength > 0 {
		objectPatcher.maxChunkLen = prm.MaxChunkLength
	} else {
		objectPatcher.maxChunkLen = defaultGRPCPayloadChunkLen
	}

	objectPatcher.req.SetBody(&v2object.PatchRequestBody{})

	meta := new(v2session.RequestMetaHeader)
	writeXHeadersToMeta(prm.XHeaders, meta)

	if prm.BearerToken != nil {
		v2BearerToken := new(acl.BearerToken)
		prm.BearerToken.WriteToV2(v2BearerToken)
		meta.SetBearerToken(v2BearerToken)
	}

	if prm.Session != nil {
		v2SessionToken := new(v2session.Token)
		prm.Session.WriteToV2(v2SessionToken)
		meta.SetSessionToken(v2SessionToken)
	}

	c.prepareRequest(&objectPatcher.req, meta)

	return &objectPatcher, nil
}

type objectPatcher struct {
	client *Client

	stream interface {
		Write(*v2object.PatchRequest) error
		Close() error
	}

	key *ecdsa.PrivateKey
	res ResObjectPatch
	err error

	addr oid.Address

	req    v2object.PatchRequest
	respV2 v2object.PatchResponse

	maxChunkLen int
}

func (x *objectPatcher) PatchAttributes(_ context.Context, newAttrs []object.Attribute, replace bool) bool {
	return x.patch(&object.Patch{
		Address:           x.addr,
		NewAttributes:     newAttrs,
		ReplaceAttributes: replace,
	})
}

func (x *objectPatcher) PatchPayload(_ context.Context, rng *object.Range, payloadReader io.Reader) bool {
	offset := rng.GetOffset()

	buf := make([]byte, x.maxChunkLen)

	for patchIter := 0; ; patchIter++ {
		n, err := payloadReader.Read(buf)
		if err != nil && err != io.EOF {
			x.err = fmt.Errorf("read payload: %w", err)
			return false
		}
		if n == 0 {
			if patchIter == 0 {
				if rng.GetLength() == 0 {
					x.err = errors.New("zero-length empty payload patch can't be applied")
					return false
				}
				if !x.patch(&object.Patch{
					Address: x.addr,
					PayloadPatch: &object.PayloadPatch{
						Range: rng,
						Chunk: []byte{},
					},
				}) {
					return false
				}
			}
			break
		}

		rngPart := object.NewRange()
		if patchIter == 0 {
			rngPart.SetOffset(offset)
			rngPart.SetLength(rng.GetLength())
		} else {
			rngPart.SetOffset(offset + rng.GetLength())
		}

		if !x.patch(&object.Patch{
			Address: x.addr,
			PayloadPatch: &object.PayloadPatch{
				Range: rngPart,
				Chunk: buf[:n],
			},
		}) {
			return false
		}

		if err == io.EOF {
			break
		}
	}

	return true
}

func (x *objectPatcher) patch(patch *object.Patch) bool {
	x.req.SetBody(patch.ToV2())
	x.req.SetVerificationHeader(nil)

	x.err = signature.SignServiceMessage(x.key, &x.req)
	if x.err != nil {
		x.err = fmt.Errorf("sign message: %w", x.err)
		return false
	}

	x.err = x.stream.Write(&x.req)
	return x.err == nil
}

func (x *objectPatcher) Close(_ context.Context) (*ResObjectPatch, error) {
	// Ignore io.EOF error, because it is expected error for client-side
	// stream termination by the server. E.g. when stream contains invalid
	// message. Server returns an error in response message (in status).
	if x.err != nil && !errors.Is(x.err, io.EOF) {
		return nil, x.err
	}

	if x.err = x.stream.Close(); x.err != nil {
		return nil, x.err
	}

	x.res.st, x.err = x.client.processResponse(&x.respV2)
	if x.err != nil || !apistatus.IsSuccessful(x.res.st) {
		return &x.res, x.err
	}

	const fieldID = "ID"

	idV2 := x.respV2.Body.ObjectID
	if idV2 == nil {
		return nil, newErrMissingResponseField(fieldID)
	}

	x.err = x.res.obj.ReadFromV2(*idV2)
	if x.err != nil {
		x.err = newErrInvalidResponseField(fieldID, x.err)
	}

	return &x.res, nil
}