frostfs-sdk-go/client/object_patch.go

package client

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

	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/acl"
	v2object "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
	rpcapi "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/client"
	v2session "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/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"
)

type ResObjectPatch struct {
	statusRes

	obj oid.ID
}

func (r ResObjectPatch) ObjectID() oid.ID {
	return r.obj
}

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

	BearerToken *bearer.Token

	Session *session.Object

	Local bool

	Address oid.Address

	Key *ecdsa.PrivateKey

	Offset uint64

	Length uint64

	Chunk []byte
}

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

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

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

	p.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)
	}

	if prm.Local {
		meta.SetTTL(1)
	}

	c.prepareRequest(&p.req, meta)
	return &p, 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
}

func (x *objectPatcher) Patch(_ context.Context, 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 {
		return nil, x.err
	}

	if !apistatus.IsSuccessful(x.res.st) {
		return &x.res, nil
	}

	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
}