frostfs-sdk-go/client/object_get.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"
	v2refs "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/refs"
	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"
	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	"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"
)

// PrmObjectGet groups parameters of ObjectGetInit operation.
type PrmObjectGet struct {
	XHeaders []string

	BearerToken *bearer.Token

	Session *session.Object

	Raw bool

	Local bool

	ContainerID *cid.ID

	ObjectID *oid.ID

	Key *ecdsa.PrivateKey
}

func (prm *PrmObjectGet) buildRequest(c *Client) (*v2object.GetRequest, error) {
	if prm.ContainerID == nil {
		return nil, errorMissingContainer
	}

	if prm.ObjectID == nil {
		return nil, errorMissingObject
	}

	if len(prm.XHeaders)%2 != 0 {
		return nil, errorInvalidXHeaders
	}

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

	addr := new(v2refs.Address)

	cnrV2 := new(v2refs.ContainerID)
	prm.ContainerID.WriteToV2(cnrV2)
	addr.SetContainerID(cnrV2)

	objV2 := new(v2refs.ObjectID)
	prm.ObjectID.WriteToV2(objV2)
	addr.SetObjectID(objV2)

	body := new(v2object.GetRequestBody)
	body.SetRaw(prm.Raw)
	body.SetAddress(addr)

	req := new(v2object.GetRequest)
	req.SetBody(body)
	c.prepareRequest(req, meta)

	return req, nil
}

// ResObjectGet groups the final result values of ObjectGetInit operation.
type ResObjectGet struct {
	statusRes
}

// ObjectReader is designed to read one object from FrostFS system.
//
// Must be initialized using Client.ObjectGetInit, any other
// usage is unsafe.
type ObjectReader struct {
	cancelCtxStream context.CancelFunc

	client *Client
	stream interface {
		Read(resp *v2object.GetResponse) error
	}

	res ResObjectGet
	err error

	tailPayload []byte

	remainingPayloadLen int
}

// UseKey specifies private key to sign the requests.
// If key is not provided, then Client default key is used.
//
// Deprecated: Use PrmObjectGet.Key instead.
func (prm *PrmObjectGet) UseKey(key ecdsa.PrivateKey) {
	prm.Key = &key
}

// ReadHeader reads header of the object. Result means success.
// Failure reason can be received via Close.
func (x *ObjectReader) ReadHeader(dst *object.Object) bool {
	var resp v2object.GetResponse
	x.err = x.stream.Read(&resp)
	if x.err != nil {
		return false
	}

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

	var partInit *v2object.GetObjectPartInit

	switch v := resp.GetBody().GetObjectPart().(type) {
	default:
		x.err = fmt.Errorf("unexpected message instead of heading part: %T", v)
		return false
	case *v2object.SplitInfo:
		x.err = object.NewSplitInfoError(object.NewSplitInfoFromV2(v))
		return false
	case *v2object.ECInfo:
		x.err = object.NewECInfoError(object.NewECInfoFromV2(v))
		return false
	case *v2object.GetObjectPartInit:
		partInit = v
	}

	var objv2 v2object.Object

	objv2.SetObjectID(partInit.GetObjectID())
	objv2.SetHeader(partInit.GetHeader())
	objv2.SetSignature(partInit.GetSignature())

	x.remainingPayloadLen = int(objv2.GetHeader().GetPayloadLength())

	*dst = *object.NewFromV2(&objv2) // need smth better

	return true
}

func (x *ObjectReader) readChunk(buf []byte) (int, bool) {
	var read int

	// read remaining tail
	read = copy(buf, x.tailPayload)

	x.tailPayload = x.tailPayload[read:]

	if len(buf) == read {
		return read, true
	}

	var chunk []byte
	var lastRead int

	for {
		var resp v2object.GetResponse
		x.err = x.stream.Read(&resp)
		if x.err != nil {
			return read, false
		}

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

		part := resp.GetBody().GetObjectPart()
		partChunk, ok := part.(*v2object.GetObjectPartChunk)
		if !ok {
			x.err = fmt.Errorf("unexpected message instead of chunk part: %T", part)
			return read, false
		}

		// read new chunk
		chunk = partChunk.GetChunk()
		if len(chunk) == 0 {
			// just skip empty chunks since they are not prohibited by protocol
			continue
		}

		lastRead = copy(buf[read:], chunk)

		read += lastRead

		if read == len(buf) {
			// save the tail
			x.tailPayload = append(x.tailPayload, chunk[lastRead:]...)

			return read, true
		}
	}
}

// ReadChunk reads another chunk of the object payload. Works similar to
// io.Reader.Read but returns success flag instead of error.
//
// Failure reason can be received via Close.
func (x *ObjectReader) ReadChunk(buf []byte) (int, bool) {
	return x.readChunk(buf)
}

func (x *ObjectReader) close(ignoreEOF bool) (*ResObjectGet, error) {
	defer x.cancelCtxStream()

	if x.err != nil {
		if !errors.Is(x.err, io.EOF) {
			return nil, x.err
		} else if !ignoreEOF {
			if x.remainingPayloadLen > 0 {
				return nil, io.ErrUnexpectedEOF
			}

			return nil, io.EOF
		}
	}

	return &x.res, nil
}

// Close ends reading the object and returns the result of the operation
// along with the final results. Must be called after using the ObjectReader.
//
// 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 errors:
//
//	*object.SplitInfoError (returned on virtual objects with PrmObjectGet.MakeRaw).
//
// Return statuses:
//   - global (see Client docs);
//   - *apistatus.ContainerNotFound;
//   - *apistatus.ObjectNotFound;
//   - *apistatus.ObjectAccessDenied;
//   - *apistatus.ObjectAlreadyRemoved;
//   - *apistatus.SessionTokenExpired.
func (x *ObjectReader) Close() (*ResObjectGet, error) {
	return x.close(true)
}

// Read implements io.Reader of the object payload.
func (x *ObjectReader) Read(p []byte) (int, error) {
	n, ok := x.readChunk(p)

	x.remainingPayloadLen -= n

	if !ok {
		res, err := x.close(false)
		if err != nil {
			return n, err
		}

		return n, apistatus.ErrFromStatus(res.Status())
	}

	if x.remainingPayloadLen < 0 {
		return n, errors.New("payload size overflow")
	}

	return n, nil
}

// ObjectGetInit initiates reading an object through a remote server using FrostFS API protocol.
//
// The call only opens the transmission channel, explicit fetching is done using the ObjectReader.
// Exactly one return value is non-nil. Resulting reader must be finally closed.
//
// Returns an error if parameters are set incorrectly (see PrmObjectGet docs).
// Context is required and must not be nil. It is used for network communication.
func (c *Client) ObjectGetInit(ctx context.Context, prm PrmObjectGet) (*ObjectReader, error) {
	req, err := prm.buildRequest(c)
	if err != nil {
		return nil, err
	}

	key := prm.Key
	if key == nil {
		key = &c.prm.Key
	}

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

	ctx, cancel := context.WithCancel(ctx)

	stream, err := rpcapi.GetObject(&c.c, req, client.WithContext(ctx))
	if err != nil {
		cancel()
		return nil, fmt.Errorf("open stream: %w", err)
	}

	var r ObjectReader
	r.cancelCtxStream = cancel
	r.stream = stream
	r.client = c

	return &r, nil
}

// PrmObjectHead groups parameters of ObjectHead operation.
type PrmObjectHead struct {
	XHeaders []string

	BearerToken *bearer.Token

	Session *session.Object

	Raw bool

	Local bool

	ContainerID *cid.ID

	ObjectID *oid.ID

	Key *ecdsa.PrivateKey
}

// UseKey specifies private key to sign the requests.
// If key is not provided, then Client default key is used.
//
// Deprecated: Use PrmObjectHead.Key instead.
func (prm *PrmObjectHead) UseKey(key ecdsa.PrivateKey) {
	prm.Key = &key
}

// ResObjectHead groups resulting values of ObjectHead operation.
type ResObjectHead struct {
	statusRes

	// requested object (response doesn't carry the ID)
	idObj oid.ID

	hdr *v2object.HeaderWithSignature
}

// ReadHeader reads header of the requested object.
// Returns false if header is missing in the response (not read).
func (x *ResObjectHead) ReadHeader(dst *object.Object) bool {
	if x.hdr == nil {
		return false
	}

	var objv2 v2object.Object

	objv2.SetHeader(x.hdr.GetHeader())
	objv2.SetSignature(x.hdr.GetSignature())

	obj := object.NewFromV2(&objv2)
	obj.SetID(x.idObj)

	*dst = *obj

	return true
}

func (prm *PrmObjectHead) buildRequest(c *Client) (*v2object.HeadRequest, error) {
	if prm.ContainerID == nil {
		return nil, errorMissingContainer
	}

	if prm.ObjectID == nil {
		return nil, errorMissingObject
	}

	if len(prm.XHeaders)%2 != 0 {
		return nil, errorInvalidXHeaders
	}

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

	addr := new(v2refs.Address)

	cnrV2 := new(v2refs.ContainerID)
	prm.ContainerID.WriteToV2(cnrV2)
	addr.SetContainerID(cnrV2)

	objV2 := new(v2refs.ObjectID)
	prm.ObjectID.WriteToV2(objV2)
	addr.SetObjectID(objV2)
	body := new(v2object.HeadRequestBody)
	body.SetRaw(prm.Raw)
	body.SetAddress(addr)

	req := new(v2object.HeadRequest)
	req.SetBody(body)
	c.prepareRequest(req, meta)

	return req, nil
}

// ObjectHead reads object header through a remote server using FrostFS API protocol.
//
// 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 `error`,
// If PrmInit.DisableFrostFSFailuresResolution has been called, unsuccessful
// FrostFS status codes are included in the returned result structure,
// otherwise, are also returned as `error`.
//
// Returns an error if parameters are set incorrectly (see PrmObjectHead docs).
// Context is required and must not be nil. It is used for network communication.
//
// Return errors:
//
//	*object.SplitInfoError (returned on virtual objects with PrmObjectHead.MakeRaw).
//
// Return statuses:
//   - global (see Client docs);
//   - *apistatus.ContainerNotFound;
//   - *apistatus.ObjectNotFound;
//   - *apistatus.ObjectAccessDenied;
//   - *apistatus.ObjectAlreadyRemoved;
//   - *apistatus.SessionTokenExpired.
func (c *Client) ObjectHead(ctx context.Context, prm PrmObjectHead) (*ResObjectHead, error) {
	req, err := prm.buildRequest(c)
	if err != nil {
		return nil, err
	}

	key := c.prm.Key
	if prm.Key != nil {
		key = *prm.Key
	}

	// sign the request

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

	resp, err := rpcapi.HeadObject(&c.c, req, client.WithContext(ctx))
	if err != nil {
		return nil, fmt.Errorf("write request: %w", err)
	}

	var res ResObjectHead
	res.st, err = c.processResponse(resp)
	if err != nil {
		return nil, err
	}

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

	res.idObj = *prm.ObjectID

	switch v := resp.GetBody().GetHeaderPart().(type) {
	default:
		return nil, fmt.Errorf("unexpected header type %T", v)
	case *v2object.SplitInfo:
		return nil, object.NewSplitInfoError(object.NewSplitInfoFromV2(v))
	case *v2object.ECInfo:
		return nil, object.NewECInfoError(object.NewECInfoFromV2(v))
	case *v2object.HeaderWithSignature:
		res.hdr = v
	}

	return &res, nil
}

// PrmObjectRange groups parameters of ObjectRange operation.
type PrmObjectRange struct {
	XHeaders []string

	BearerToken *bearer.Token

	Session *session.Object

	Raw bool

	Local bool

	ContainerID *cid.ID

	ObjectID *oid.ID

	Key *ecdsa.PrivateKey

	Offset uint64

	Length uint64
}

func (prm *PrmObjectRange) buildRequest(c *Client) (*v2object.GetRangeRequest, error) {
	if prm.Length == 0 {
		return nil, errorZeroRangeLength
	}

	if prm.ContainerID == nil {
		return nil, errorMissingContainer
	}

	if prm.ObjectID == nil {
		return nil, errorMissingObject
	}

	if len(prm.XHeaders)%2 != 0 {
		return nil, errorInvalidXHeaders
	}

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

	addr := new(v2refs.Address)

	cnrV2 := new(v2refs.ContainerID)
	prm.ContainerID.WriteToV2(cnrV2)
	addr.SetContainerID(cnrV2)

	objV2 := new(v2refs.ObjectID)
	prm.ObjectID.WriteToV2(objV2)
	addr.SetObjectID(objV2)

	rng := new(v2object.Range)
	rng.SetLength(prm.Length)
	rng.SetOffset(prm.Offset)

	body := new(v2object.GetRangeRequestBody)
	body.SetRaw(prm.Raw)
	body.SetAddress(addr)
	body.SetRange(rng)

	req := new(v2object.GetRangeRequest)
	req.SetBody(body)
	c.prepareRequest(req, meta)

	return req, nil
}

// UseKey specifies private key to sign the requests.
// If key is not provided, then Client default key is used.
//
// Deprecated: Use PrmObjectRange.Key instead.
func (prm *PrmObjectRange) UseKey(key ecdsa.PrivateKey) {
	prm.Key = &key
}

// ResObjectRange groups the final result values of ObjectRange operation.
type ResObjectRange struct {
	statusRes
}

// ObjectRangeReader is designed to read payload range of one object
// from FrostFS system.
//
// Must be initialized using Client.ObjectRangeInit, any other
// usage is unsafe.
type ObjectRangeReader struct {
	cancelCtxStream context.CancelFunc

	client *Client

	res ResObjectRange
	err error

	stream interface {
		Read(resp *v2object.GetRangeResponse) error
	}

	tailPayload []byte

	remainingPayloadLen int
}

func (x *ObjectRangeReader) readChunk(buf []byte) (int, bool) {
	var read int

	// read remaining tail
	read = copy(buf, x.tailPayload)

	x.tailPayload = x.tailPayload[read:]

	if len(buf) == read {
		return read, true
	}

	var partChunk *v2object.GetRangePartChunk
	var chunk []byte
	var lastRead int

	for {
		var resp v2object.GetRangeResponse
		x.err = x.stream.Read(&resp)
		if x.err != nil {
			return read, false
		}

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

		// get chunk message
		switch v := resp.GetBody().GetRangePart().(type) {
		default:
			x.err = fmt.Errorf("unexpected message received: %T", v)
			return read, false
		case *v2object.SplitInfo:
			x.err = object.NewSplitInfoError(object.NewSplitInfoFromV2(v))
			return read, false
		case *v2object.GetRangePartChunk:
			partChunk = v
		}

		chunk = partChunk.GetChunk()
		if len(chunk) == 0 {
			// just skip empty chunks since they are not prohibited by protocol
			continue
		}

		lastRead = copy(buf[read:], chunk)

		read += lastRead

		if read == len(buf) {
			// save the tail
			x.tailPayload = append(x.tailPayload, chunk[lastRead:]...)

			return read, true
		}
	}
}

// ReadChunk reads another chunk of the object payload range.
// Works similar to io.Reader.Read but returns success flag instead of error.
//
// Failure reason can be received via Close.
func (x *ObjectRangeReader) ReadChunk(buf []byte) (int, bool) {
	return x.readChunk(buf)
}

func (x *ObjectRangeReader) close(ignoreEOF bool) (*ResObjectRange, error) {
	defer x.cancelCtxStream()

	if x.err != nil {
		if !errors.Is(x.err, io.EOF) {
			return nil, x.err
		} else if !ignoreEOF {
			if x.remainingPayloadLen > 0 {
				return nil, io.ErrUnexpectedEOF
			}

			return nil, io.EOF
		}
	}

	return &x.res, nil
}

// Close ends reading the payload range and returns the result of the operation
// along with the final results. Must be called after using the ObjectRangeReader.
//
// 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 errors:
//
//	*object.SplitInfoError (returned on virtual objects with PrmObjectRange.MakeRaw).
//
// Return statuses:
//   - global (see Client docs);
//   - *apistatus.ContainerNotFound;
//   - *apistatus.ObjectNotFound;
//   - *apistatus.ObjectAccessDenied;
//   - *apistatus.ObjectAlreadyRemoved;
//   - *apistatus.ObjectOutOfRange;
//   - *apistatus.SessionTokenExpired.
func (x *ObjectRangeReader) Close() (*ResObjectRange, error) {
	return x.close(true)
}

// Read implements io.Reader of the object payload.
func (x *ObjectRangeReader) Read(p []byte) (int, error) {
	n, ok := x.readChunk(p)

	x.remainingPayloadLen -= n

	if !ok {
		res, err := x.close(false)
		if err != nil {
			return n, err
		}

		return n, apistatus.ErrFromStatus(res.Status())
	}

	if x.remainingPayloadLen < 0 {
		return n, errors.New("payload range size overflow")
	}

	return n, nil
}

// ObjectRangeInit initiates reading an object's payload range through a remote
// server using FrostFS API protocol.
//
// The call only opens the transmission channel, explicit fetching is done using the ObjectRangeReader.
// Exactly one return value is non-nil. Resulting reader must be finally closed.
//
// Returns an error if parameters are set incorrectly (see PrmObjectRange docs).
// Context is required and must not be nil. It is used for network communication.
func (c *Client) ObjectRangeInit(ctx context.Context, prm PrmObjectRange) (*ObjectRangeReader, error) {
	req, err := prm.buildRequest(c)
	if err != nil {
		return nil, err
	}

	key := prm.Key
	if key == nil {
		key = &c.prm.Key
	}

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

	ctx, cancel := context.WithCancel(ctx)

	stream, err := rpcapi.GetObjectRange(&c.c, req, client.WithContext(ctx))
	if err != nil {
		cancel()
		return nil, fmt.Errorf("open stream: %w", err)
	}

	var r ObjectRangeReader
	r.remainingPayloadLen = int(prm.Length)
	r.cancelCtxStream = cancel
	r.stream = stream
	r.client = c

	return &r, nil
}