feat: reserve event types (#3679)

Author: chrisgacsal

app/common/meter.go

@@ -2,8 +2,10 @@ package common
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
+	"regexp"
 
 	"github.com/google/wire"
 	"github.com/samber/lo"
@@ -27,13 +29,15 @@ var Meter = wire.NewSet(
 var MeterManage = wire.NewSet(
 	Meter,
 	NewMeterManageService,
+	NewReservedEventTypePatterns,
 )
 
 var MeterManageWithConfigMeters = wire.NewSet(
 	wire.FieldsOf(new(config.Configuration), "Meters"),
 
 	Meter,
 	NewMeterManageService,
+	NewReservedEventTypePatterns,
 	NewMeterConfigInitializer,
 )
 
@@ -43,15 +47,36 @@ func NewMeterService(
 	return service.New(meterAdapter)
 }
 
+func NewReservedEventTypePatterns(reserved []config.ReservedEventTypePattern) ([]*meter.EventTypePattern, error) {
+	var errs []error
+
+	patterns := make([]*meter.EventTypePattern, 0, len(reserved))
+
+	for _, r := range reserved {
+		pattern, err := regexp.Compile(r)
+		if err != nil {
+			errs = append(errs, fmt.Errorf("invalid reserved event type pattern %q: %w", r, err))
+
+			continue
+		}
+
+		patterns = append(patterns, pattern)
+	}
+
+	return patterns, errors.Join(errs...)
+}
+
 func NewMeterManageService(
 	meterAdapter *adapter.Adapter,
 	namespaceManager *namespace.Manager,
 	publisher eventbus.Publisher,
+	reservedEventTypes []*meter.EventTypePattern,
 ) meter.ManageService {
 	return service.NewManage(
 		meterAdapter,
 		publisher,
 		namespaceManager,
+		reservedEventTypes,
 	)
 }
 

app/config/config.go

@@ -3,6 +3,8 @@ package config
 
 import (
 	"errors"
+	"fmt"
+	"regexp"
 	"strings"
 	"time"
 
@@ -15,6 +17,8 @@ import (
 	"github.com/openmeterio/openmeter/pkg/models"
 )
 
+type ReservedEventTypePattern = string
+
 // Configuration holds any kind of Configuration that comes from the outside world and
 // is necessary for running the application.
 type Configuration struct {
@@ -25,24 +29,25 @@ type Configuration struct {
 
 	Termination TerminationConfig
 
-	Aggregation     AggregationConfiguration
-	Entitlements    EntitlementsConfiguration
-	Customer        CustomerConfiguration
-	Dedupe          DedupeConfiguration
-	Events          EventsConfiguration
-	Ingest          IngestConfiguration
-	Meters          []*meter.Meter
-	Namespace       NamespaceConfiguration
-	Portal          PortalConfiguration
-	Postgres        PostgresConfig
-	Sink            SinkConfiguration
-	BalanceWorker   BalanceWorkerConfiguration
-	Notification    NotificationConfiguration
-	ProductCatalog  ProductCatalogConfiguration
-	ProgressManager ProgressManagerConfiguration
-	Billing         BillingConfiguration
-	Apps            AppsConfiguration
-	Svix            SvixConfig
+	Aggregation        AggregationConfiguration
+	Entitlements       EntitlementsConfiguration
+	Customer           CustomerConfiguration
+	Dedupe             DedupeConfiguration
+	Events             EventsConfiguration
+	Ingest             IngestConfiguration
+	Meters             []*meter.Meter
+	ReservedEventTypes []ReservedEventTypePattern
+	Namespace          NamespaceConfiguration
+	Portal             PortalConfiguration
+	Postgres           PostgresConfig
+	Sink               SinkConfiguration
+	BalanceWorker      BalanceWorkerConfiguration
+	Notification       NotificationConfiguration
+	ProductCatalog     ProductCatalogConfiguration
+	ProgressManager    ProgressManagerConfiguration
+	Billing            BillingConfiguration
+	Apps               AppsConfiguration
+	Svix               SvixConfig
 }
 
 // Validate validates the configuration.
@@ -107,6 +112,12 @@ func (c Configuration) Validate() error {
 		}
 	}
 
+	for _, pattern := range c.ReservedEventTypes {
+		if _, err := regexp.Compile(pattern); err != nil {
+			errs = append(errs, fmt.Errorf("reserved event type pattern %q: invalid regular expression", pattern))
+		}
+	}
+
 	if err := c.BalanceWorker.Validate(); err != nil {
 		errs = append(errs, errorsx.WithPrefix(err, "balance worker"))
 	}

app/config/config_test.go

@@ -411,6 +411,11 @@ func TestComplete(t *testing.T) {
 			EnableSubjectHook: true,
 			IgnoreErrors:      true,
 		},
+		ReservedEventTypes: []string{
+			`^reserved\..*$`,
+			`^_\..*$`,
+			`^openmeter\..*$`,
+		},
 	}
 
 	assert.Equal(t, expected, actual)

app/config/testdata/complete.yaml

@@ -139,6 +139,11 @@ meters:
       method: $.method
       path: $.path
 
+reservedEventTypes:
+  - ^reserved\..*$
+  - ^_\..*$
+  - ^openmeter\..*$
+
 notification:
   enabled: true
   reconcileInterval: 1m

cmd/server/wire.go

@@ -86,6 +86,8 @@ type Application struct {
 
 func initializeApplication(ctx context.Context, conf config.Configuration) (Application, func(), error) {
 	wire.Build(
+		wire.FieldsOf(new(config.Configuration), "ReservedEventTypes"),
+
 		metadata,
 		common.App,
 		common.Billing,

cmd/server/wire_gen.go

@@ -21,7 +21,7 @@ import (
 // CreateMeter creates a new meter.
 func (a *Adapter) CreateMeter(ctx context.Context, input meterpkg.CreateMeterInput) (meterpkg.Meter, error) {
 	if err := input.Validate(); err != nil {
-		return meterpkg.Meter{}, models.NewGenericValidationError(err)
+		return meterpkg.Meter{}, err
 	}
 
 	return transaction.Run(ctx, a, func(ctx context.Context) (meterpkg.Meter, error) {

openmeter/meter/adapter/manage.go

@@ -12,6 +12,8 @@ import (
 
 var groupByKeyRegExp = regexp.MustCompile(`^[a-zA-Z_][0-9a-zA-Z_]*$`)
 
+type EventTypePattern = regexp.Regexp
+
 type MeterAggregation string
 
 // Note: keep values up to date in the meter package

openmeter/meter/meter.go

@@ -103,6 +103,15 @@ func (p ListMetersParams) Validate() error {
 	return errors.Join(errs...)
 }
 
+type inputOptions struct {
+	AllowReservedEventTypes bool
+}
+
+var (
+	_ models.Validator                         = (*CreateMeterInput)(nil)
+	_ models.CustomValidator[CreateMeterInput] = (*CreateMeterInput)(nil)
+)
+
 // CreateMeterInput is a parameter object for creating a meter.
 type CreateMeterInput struct {
 	Namespace     string
@@ -116,6 +125,32 @@ type CreateMeterInput struct {
 	GroupBy       map[string]string
 	Metadata      models.Metadata
 	Annotations   models.Annotations
+
+	inputOptions
+}
+
+func (i CreateMeterInput) ValidateWith(validators ...models.ValidatorFunc[CreateMeterInput]) error {
+	return models.Validate(i, validators...)
+}
+
+func ValidateCreateMeterInputWithReservedEventTypes(reserved []*EventTypePattern) models.ValidatorFunc[CreateMeterInput] {
+	return func(input CreateMeterInput) error {
+		if input.AllowReservedEventTypes {
+			return nil
+		}
+
+		for _, pattern := range reserved {
+			if pattern == nil {
+				continue
+			}
+
+			if ok := pattern.MatchString(input.EventType); ok {
+				return fmt.Errorf("event type '%s' is reserved: matched pattern '%s'", input.EventType, pattern.String())
+			}
+		}
+
+		return nil
+	}
 }
 
 // Validate validates the create meter input.
@@ -160,7 +195,7 @@ func (i CreateMeterInput) Validate() error {
 		errs = append(errs, fmt.Errorf("invalid meter group by: %w", err))
 	}
 
-	return errors.Join(errs...)
+	return models.NewNillableGenericValidationError(errors.Join(errs...))
 }
 
 // UpdateMeterInput is a parameter object for creating a meter.

openmeter/meter/service.go

@@ -17,22 +17,25 @@ var _ meter.ManageService = (*ManageService)(nil)
 
 type ManageService struct {
 	meter.Service
-	preUpdateHooks   []meter.PreUpdateMeterHook
-	adapter          *adapter.Adapter
-	publisher        eventbus.Publisher
-	namespaceManager *namespace.Manager
+	preUpdateHooks     []meter.PreUpdateMeterHook
+	adapter            *adapter.Adapter
+	publisher          eventbus.Publisher
+	namespaceManager   *namespace.Manager
+	reservedEventTypes []*meter.EventTypePattern
 }
 
 func NewManage(
 	adapter *adapter.Adapter,
 	publisher eventbus.Publisher,
 	namespaceManager *namespace.Manager,
+	reservedEventTypes []*meter.EventTypePattern,
 ) *ManageService {
 	return &ManageService{
-		Service:          New(adapter),
-		adapter:          adapter,
-		publisher:        publisher,
-		namespaceManager: namespaceManager,
+		Service:            New(adapter),
+		adapter:            adapter,
+		publisher:          publisher,
+		namespaceManager:   namespaceManager,
+		reservedEventTypes: reservedEventTypes,
 	}
 }
 
@@ -44,6 +47,17 @@ func (s *ManageService) RegisterPreUpdateMeterHook(hook meter.PreUpdateMeterHook
 
 // CreateMeter creates a meter
 func (s *ManageService) CreateMeter(ctx context.Context, input meter.CreateMeterInput) (meter.Meter, error) {
+	if err := input.Validate(); err != nil {
+		return meter.Meter{}, fmt.Errorf("invalid create meter params: %w", err)
+	}
+
+	if err := input.ValidateWith(
+		// Validate with reserved event types
+		meter.ValidateCreateMeterInputWithReservedEventTypes(s.reservedEventTypes),
+	); err != nil {
+		return meter.Meter{}, fmt.Errorf("invalid create meter params: %w", err)
+	}
+
 	// Create the meter
 	createdMeter, err := s.adapter.CreateMeter(ctx, input)
 	if err != nil {