Method ReadAttributes

internal/storage/memory/data_reader.go:274–328

ReadAttributes reads attributes from the database taking into account the pagination.

(_ context.Context, tenantID string, filter *base.AttributeFilter, _ string, pagination database.Pagination)

// ReadAttributes reads attributes from the database taking into account the pagination.
func (r *DataReader) ReadAttributes(_ context.Context, tenantID string, filter *base.AttributeFilter, _ string, pagination database.Pagination) (collection *database.AttributeCollection, ct database.EncodedContinuousToken, err error) {
	txn := r.database.DB.Txn(false)
	defer txn.Abort()

	var lowerBound uint64
	if pagination.Token() != "" {
		var t database.ContinuousToken
		t, err = utils.EncodedContinuousToken{Value: pagination.Token()}.Decode()
		if err != nil {
			return nil, database.NewNoopContinuousToken().Encode(), err
		}
		lowerBound, err = strconv.ParseUint(t.(utils.ContinuousToken).Value, 10, 64)
		if err != nil {
			return nil, database.NewNoopContinuousToken().Encode(), errors.New(base.ErrorCode_ERROR_CODE_INVALID_CONTINUOUS_TOKEN.String())
		}
	}

	// Get the index and arguments based on the filter.
	index, args := utils.GetAttributesIndexNameAndArgsByFilters(tenantID, filter)

	// Get the result iterator using lower bound.
	var result memdb.ResultIterator
	result, err = txn.LowerBound(constants.AttributesTable, index, args...)
	if err != nil {
		return nil, database.NewNoopContinuousToken().Encode(), errors.New(base.ErrorCode_ERROR_CODE_EXECUTION.String())
	}

	// Filter the result iterator and add the attributes to the array.
	attr := make([]storage.Attribute, 0, 10)
	fit := memdb.NewFilterIterator(result, utils.FilterAttributesQuery(tenantID, filter))
	for obj := fit.Next(); obj != nil; obj = fit.Next() {
		a, ok := obj.(storage.Attribute)
		if !ok {
			return nil, database.NewNoopContinuousToken().Encode(), errors.New(base.ErrorCode_ERROR_CODE_TYPE_CONVERSATION.String())
		}
		attr = append(attr, a)
	}

	// Sort the attributes and append them to the collection.
	sort.Slice(attr, func(i, j int) bool {
		return attr[i].ID < attr[j].ID
	})

	attributes := make([]*base.Attribute, 0, pagination.PageSize()+1)
	for _, t := range attr {
		if t.ID >= lowerBound {
			attributes = append(attributes, t.ToAttribute())
			if pagination.PageSize() != 0 && len(attributes) > int(pagination.PageSize()) {
				return database.NewAttributeCollection(attributes[:pagination.PageSize()]...), utils.NewContinuousToken(strconv.FormatUint(t.ID, 10)).Encode(), nil
			}
		}
	}

	return database.NewAttributeCollection(attributes...), database.NewNoopContinuousToken().Encode(), nil
}

// QueryUniqueSubjectReferences is a function that searches for unique subject references in a given database.
func (r *DataReader) QueryUniqueSubjectReferences(_ context.Context, tenantID string, subjectReference *base.RelationReference, excluded []string, _ string, pagination database.Pagination) (ids []string, _ database.EncodedContinuousToken, err error) {