streaming/connector.go

package streaming

import (
	"context"
	"errors"
	"time"

	"github.com/openmeterio/openmeter/openmeter/meter"
	"github.com/openmeterio/openmeter/openmeter/namespace"
	"github.com/openmeterio/openmeter/pkg/models"
)

type CountEventsParams struct {
	From time.Time
}

// CountEventRow represents a row in the count events response.
type CountEventRow struct {
	Count   uint64
	Subject string
}

// RawEvent represents a single raw event
type RawEvent struct {
	Namespace  string    `ch:"namespace" json:"-"`
	ID         string    `ch:"id" json:"id"`
	Type       string    `ch:"type" json:"type"`
	Source     string    `ch:"source" json:"source"`
	Subject    string    `ch:"subject" json:"subject"`
	Time       time.Time `ch:"time" json:"time"`
	Data       string    `ch:"data" json:"data"`
	IngestedAt time.Time `ch:"ingested_at" json:"ingested_at,omitempty,omitzero"`
	StoredAt   time.Time `ch:"stored_at" json:"stored_at,omitempty,omitzero"`
	StoreRowID string    `ch:"store_row_id" json:"store_row_id,omitempty,omitzero"`
	CustomerID *string   `ch:"customer_id" json:"customer_id,omitempty,omitzero"`
}

type Connector interface {
	namespace.Handler

	CountEvents(ctx context.Context, namespace string, params CountEventsParams) ([]CountEventRow, error)
	ListEvents(ctx context.Context, namespace string, params ListEventsParams) ([]RawEvent, error)
	ListEventsV2(ctx context.Context, params ListEventsV2Params) ([]RawEvent, error)
	// ListSubjects lists the subjects that have events in the database
	ListSubjects(ctx context.Context, params ListSubjectsParams) ([]string, error)
	// ListGroupByValues lists the group by values that have events in the database
	ListGroupByValues(ctx context.Context, params ListGroupByValuesParams) ([]string, error)
	QueryMeter(ctx context.Context, namespace string, meter meter.Meter, params QueryParams) ([]meter.MeterQueryRow, error)
	BatchInsert(ctx context.Context, events []RawEvent) error
	ValidateJSONPath(ctx context.Context, jsonPath string) (bool, error)
}

// ListSubjectsParams is a parameter object for listing subjects.
type ListSubjectsParams struct {
	Namespace string
	Meter     *meter.Meter
	From      *time.Time
	To        *time.Time
	Search    *string
}

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

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

	if p.Meter != nil {
		if p.Meter.Key == "" {
			errs = append(errs, errors.New("meter cannot be empty when provided"))
		}
	}

	if p.From != nil && p.To != nil {
		if p.From.Equal(*p.To) {
			errs = append(errs, errors.New("from and to cannot be equal"))
		}

		if p.From.After(*p.To) {
			errs = append(errs, errors.New("from time must be before to time"))
		}
	}

	return models.NewNillableGenericValidationError(errors.Join(errs...))
}

// ListGroupByValuesParams is a parameter object for listing group by values.
type ListGroupByValuesParams struct {
	Namespace  string
	Meter      meter.Meter
	GroupByKey string
	From       *time.Time
	To         *time.Time
	Search     *string
}

// Validate validates the list group by values parameters.
func (p ListGroupByValuesParams) Validate() error {
	var errs []error

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

	if p.GroupByKey == "" {
		errs = append(errs, errors.New("group by key is required"))
	}

	if p.Meter.GroupBy[p.GroupByKey] == "" {
		errs = append(errs, errors.New("group by key is not valid for this meter"))
	}

	if p.From != nil {
		if time.Since(*p.From) >= time.Hour*24*90 {
			errs = append(errs, errors.New("from time must not be more than 90 days ago"))
		}
	}

	if p.From != nil && p.To != nil {
		if p.From.Equal(*p.To) {
			errs = append(errs, errors.New("from and to cannot be equal"))
		}

		if p.From.After(*p.To) {
			errs = append(errs, errors.New("from time must be before to time"))
		}

		if p.To.Sub(*p.From) > time.Hour*24*30 {
			errs = append(errs, errors.New("time window must be less than 30 days"))
		}
	}

	return models.NewNillableGenericValidationError(errors.Join(errs...))
}