streaming/eventparams.go

package streaming

import (
	"errors"
	"fmt"
	"time"

	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/pkg/filter"
	"github.com/openmeterio/openmeter/pkg/models"
	"github.com/openmeterio/openmeter/pkg/pagination/v2"
	"github.com/openmeterio/openmeter/pkg/sortx"
)

// EventSortField is the column used to order events returned by ListEventsV2.
type EventSortField string

const (
	EventSortFieldTime       EventSortField = "time"
	EventSortFieldIngestedAt EventSortField = "ingested_at"
	EventSortFieldStoredAt   EventSortField = "stored_at"
)

// Values returns the known values for the event sort field.
func (f EventSortField) Values() []string {
	return []string{
		string(EventSortFieldTime),
		string(EventSortFieldIngestedAt),
		string(EventSortFieldStoredAt),
	}
}

// Validate returns an error when the sort field is not one of the known values.
// The zero value is treated as valid and resolved to EventSortFieldTime at query time.
func (f EventSortField) Validate() error {
	if f == "" {
		return nil
	}

	if !lo.Contains(f.Values(), string(f)) {
		return models.NewGenericValidationError(fmt.Errorf("invalid event sort value: %s", f))
	}

	return nil
}

// ListEventsParams represents the input for ListEvents method.
type ListEventsParams struct {
	// The namespace.
	Namespace string
	// The client ID.
	ClientID *string
	// Start date-time. Inclusive.
	IngestedAtFrom *time.Time
	// End date-time. Inclusive.
	IngestedAtTo *time.Time
	// The event ID. Accepts partial ID.
	ID *string
	// The event subject. Accepts partial subject.
	Subject *string
	// The event customer ID.
	Customers *[]Customer
	// Start date-time. Inclusive.
	From time.Time
	// End date-time. Inclusive.
	To *time.Time
	// Number of events to return.
	Limit int
}

// Validate validates the input.
func (i ListEventsParams) Validate() error {
	var errs []error

	if i.Namespace == "" {
		errs = append(errs, errors.New("namespace is required"))
	}

	if i.ClientID != nil && *i.ClientID == "" {
		errs = append(errs, errors.New("client id cannot be empty"))
	}

	if i.From.IsZero() {
		errs = append(errs, errors.New("from date is required"))
	}

	if i.To != nil && i.To.Before(i.From) {
		errs = append(errs, fmt.Errorf("to date is before from date: %s < %s", i.To.Format(time.RFC3339), i.From.Format(time.RFC3339)))
	}

	if i.IngestedAtFrom != nil && i.IngestedAtTo != nil && i.IngestedAtTo.Before(*i.IngestedAtFrom) {
		errs = append(errs, fmt.Errorf("ingestedAtTo date is before ingestedAtFrom date: %s < %s", i.IngestedAtTo.Format(time.RFC3339), i.IngestedAtFrom.Format(time.RFC3339)))
	}

	return errors.Join(errs...)
}

// ListEventsV2Params is a parameter object for listing events.
type ListEventsV2Params struct {
	// The namespace.
	Namespace string
	// The client ID.
	ClientID *string
	// The cursor.
	Cursor *pagination.Cursor
	// The limit.
	Limit *int
	// The ID filter.
	ID *filter.FilterString
	// The source filter.
	Source *filter.FilterString
	// The subject filter.
	Subject *filter.FilterString
	// The event customer ID.
	Customers *[]Customer
	// The type filter.
	Type *filter.FilterString
	// The time filter.
	Time *filter.FilterTime
	// The ingested at filter.
	IngestedAt *filter.FilterTime
	// The stored at filter.
	StoredAt *filter.FilterTime
	// The sort column; zero value defaults to EventSortFieldTime.
	SortBy EventSortField
	// The sort direction; zero value defaults to sortx.OrderDesc.
	SortOrder sortx.Order
}

// Validate validates the list events parameters.
func (p ListEventsV2Params) Validate() error {
	var errs []error

	if p.Namespace == "" {
		errs = append(errs, errors.New("namespace is required"))
	}

	if p.Cursor != nil {
		if err := p.Cursor.Validate(); err != nil {
			errs = append(errs, fmt.Errorf("cursor: %w", err))
		}
	}

	if p.ID != nil {
		if err := p.ID.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("id: %w", err))
		}
	}

	if p.Source != nil {
		if err := p.Source.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("source: %w", err))
		}
	}

	if p.Subject != nil {
		if err := p.Subject.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("subject: %w", err))
		}
	}

	if p.Type != nil {
		if err := p.Type.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("type: %w", err))
		}
	}

	if p.Time != nil {
		if err := p.Time.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("time: %w", err))
		}
	}

	if p.IngestedAt != nil {
		if err := p.IngestedAt.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("ingested_at: %w", err))
		}
	}

	if p.StoredAt != nil {
		if err := p.StoredAt.ValidateWithComplexity(1); err != nil {
			errs = append(errs, fmt.Errorf("stored_at: %w", err))
		}
	}

	if err := p.SortBy.Validate(); err != nil {
		errs = append(errs, fmt.Errorf("sort by: %w", err))
	}

	if p.Limit != nil && *p.Limit < 1 {
		errs = append(errs, errors.New("limit must be greater than 0"))
	}

	return errors.Join(errs...)
}