frostfs-sdk-go/client/object_search.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"
	v2refs "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/refs"
	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"
	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"
)

// PrmObjectSearch groups parameters of ObjectSearch operation.
type PrmObjectSearch struct {
	XHeaders []string

	Local bool

	BearerToken *bearer.Token

	Session *session.Object

	ContainerID *cid.ID

	Key *ecdsa.PrivateKey

	Filters object.SearchFilters
}

// MarkLocal tells the server to execute the operation locally.
//
// Deprecated: Use PrmObjectSearch.Local instead.
func (x *PrmObjectSearch) MarkLocal() {
	x.Local = true
}

// WithinSession specifies session within which the search query must be executed.
//
// Creator of the session acquires the authorship of the request.
// This may affect the execution of an operation (e.g. access control).
//
// Must be signed.
//
// Deprecated: Use PrmObjectSearch.Session instead.
func (x *PrmObjectSearch) WithinSession(t session.Object) {
	x.Session = &t
}

// WithBearerToken attaches bearer token to be used for the operation.
//
// If set, underlying eACL rules will be used in access control.
//
// Must be signed.
//
// Deprecated: Use PrmObjectSearch.BearerToken instead.
func (x *PrmObjectSearch) WithBearerToken(t bearer.Token) {
	x.BearerToken = &t
}

// WithXHeaders specifies list of extended headers (string key-value pairs)
// to be attached to the request. Must have an even length.
//
// Slice must not be mutated until the operation completes.
//
// Deprecated: Use PrmObjectSearch.XHeaders instead.
func (x *PrmObjectSearch) WithXHeaders(hs ...string) {
	x.XHeaders = hs
}

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

// InContainer specifies the container in which to look for objects.
// Required parameter.
//
// Deprecated: Use PrmObjectSearch.ContainerID instead.
func (x *PrmObjectSearch) InContainer(id cid.ID) {
	x.ContainerID = &id
}

// SetFilters sets filters by which to select objects. All container objects
// match unset/empty filters.
//
// Deprecated: Use PrmObjectSearch.Filters instead.
func (x *PrmObjectSearch) SetFilters(filters object.SearchFilters) {
	x.Filters = filters
}

// ResObjectSearch groups the final result values of ObjectSearch operation.
type ResObjectSearch struct {
	statusRes
}

// ObjectListReader is designed to read list of object identifiers from FrostFS system.
//
// Must be initialized using Client.ObjectSearch, any other usage is unsafe.
type ObjectListReader struct {
	client          *Client
	cancelCtxStream context.CancelFunc
	err             error
	res             ResObjectSearch
	stream          interface {
		Read(resp *v2object.SearchResponse) error
	}
	tail []v2refs.ObjectID
}

// Read reads another list of the object identifiers. Works similar to
// io.Reader.Read but copies oid.ID and returns success flag instead of error.
//
// Failure reason can be received via Close.
//
// Panics if buf has zero length.
func (x *ObjectListReader) Read(buf []oid.ID) (int, bool) {
	if len(buf) == 0 {
		panic("empty buffer in ObjectListReader.ReadList")
	}

	read := copyIDBuffers(buf, x.tail)
	x.tail = x.tail[read:]

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

	for {
		var resp v2object.SearchResponse
		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
		}

		// read new chunk of objects
		ids := resp.GetBody().GetIDList()
		if len(ids) == 0 {
			// just skip empty lists since they are not prohibited by protocol
			continue
		}

		ln := copyIDBuffers(buf[read:], ids)
		read += ln

		if read == len(buf) {
			// save the tail
			x.tail = append(x.tail, ids[ln:]...)

			return read, true
		}
	}
}

func copyIDBuffers(dst []oid.ID, src []v2refs.ObjectID) int {
	var i int
	for ; i < len(dst) && i < len(src); i++ {
		_ = dst[i].ReadFromV2(src[i])
	}
	return i
}

// Iterate iterates over the list of found object identifiers.
// f can return true to stop iteration earlier.
//
// Returns an error if object can't be read.
func (x *ObjectListReader) Iterate(f func(oid.ID) bool) error {
	buf := make([]oid.ID, 1)

	for {
		// Do not check first return value because `len(buf) == 1`,
		// so false means nothing was read.
		_, ok := x.Read(buf)
		if !ok {
			res, err := x.Close()
			if err != nil {
				return err
			}
			return apistatus.ErrFromStatus(res.Status())
		}
		if f(buf[0]) {
			return nil
		}
	}
}

// Close ends reading list of the matched objects and returns the result of the operation
// along with the final results. Must be called after using the ObjectListReader.
//
// 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.ObjectAccessDenied;
//   - *apistatus.SessionTokenExpired.
func (x *ObjectListReader) Close() (*ResObjectSearch, error) {
	defer x.cancelCtxStream()

	if x.err != nil && !errors.Is(x.err, io.EOF) {
		return &x.res, x.err
	}

	return &x.res, nil
}

func (x *PrmObjectSearch) buildRequest(c *Client) (*v2object.SearchRequest, error) {
	if x.ContainerID == nil {
		return nil, errorMissingContainer
	}

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

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

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

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

	if x.Local {
		meta.SetTTL(1)
	}
	cnrV2 := new(v2refs.ContainerID)
	x.ContainerID.WriteToV2(cnrV2)

	body := new(v2object.SearchRequestBody)
	body.SetVersion(1)
	body.SetContainerID(cnrV2)
	body.SetFilters(x.Filters.ToV2())

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

	return req, nil
}

// ObjectSearchInit initiates object selection through a remote server using FrostFS API protocol.
//
// The call only opens the transmission channel, explicit fetching of matched objects
// is done using the ObjectListReader. Exactly one return value is non-nil.
// Resulting reader must be finally closed.
//
// Returns an error if parameters are set incorrectly (see PrmObjectSearch docs).
// Context is required and must not be nil. It is used for network communication.
func (c *Client) ObjectSearchInit(ctx context.Context, prm PrmObjectSearch) (*ObjectListReader, 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)
	}

	var r ObjectListReader
	ctx, r.cancelCtxStream = context.WithCancel(ctx)

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

	return &r, nil
}