| 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" |
| ) |
|
|
| |
| type EventSortField string |
|
|
| const ( |
| EventSortFieldTime EventSortField = "time" |
| EventSortFieldIngestedAt EventSortField = "ingested_at" |
| EventSortFieldStoredAt EventSortField = "stored_at" |
| ) |
|
|
| |
| func (f EventSortField) Values() []string { |
| return []string{ |
| string(EventSortFieldTime), |
| string(EventSortFieldIngestedAt), |
| string(EventSortFieldStoredAt), |
| } |
| } |
|
|
| |
| |
| 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 |
| } |
|
|
| |
| type ListEventsParams struct { |
| |
| Namespace string |
| |
| ClientID *string |
| |
| IngestedAtFrom *time.Time |
| |
| IngestedAtTo *time.Time |
| |
| ID *string |
| |
| Subject *string |
| |
| Customers *[]Customer |
| |
| From time.Time |
| |
| To *time.Time |
| |
| Limit int |
| } |
|
|
| |
| 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...) |
| } |
|
|
| |
| type ListEventsV2Params struct { |
| |
| Namespace string |
| |
| ClientID *string |
| |
| Cursor *pagination.Cursor |
| |
| Limit *int |
| |
| ID *filter.FilterString |
| |
| Source *filter.FilterString |
| |
| Subject *filter.FilterString |
| |
| Customers *[]Customer |
| |
| Type *filter.FilterString |
| |
| Time *filter.FilterTime |
| |
| IngestedAt *filter.FilterTime |
| |
| StoredAt *filter.FilterTime |
| |
| SortBy EventSortField |
| |
| SortOrder sortx.Order |
| } |
|
|
| |
| 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...) |
| } |
|
|