feat: list meters by event type (#3685)

Author: chrisgacsal

openmeter/ent/db/migrate/schema.go

@@ -38,11 +38,14 @@ func (Meter) Fields() []ent.Field {
 
 // Indexes of the Meter.
 func (Meter) Indexes() []ent.Index {
-	return []ent.Index{index.Fields("namespace", "key").
-		Annotations(
-			entsql.IndexWhere("deleted_at IS NULL"),
-		).
-		Unique()}
+	return []ent.Index{
+		index.Fields("namespace", "key").
+			Annotations(
+				entsql.IndexWhere("deleted_at IS NULL"),
+			).
+			Unique(),
+		index.Fields("namespace", "event_type"),
+	}
 }
 
 // Edges of the Meter.

openmeter/ent/schema/meter.go

@@ -0,0 +1,160 @@
+package adapter
+
+import (
+	"crypto/rand"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	entdb "github.com/openmeterio/openmeter/openmeter/ent/db"
+	"github.com/openmeterio/openmeter/openmeter/meter"
+	"github.com/openmeterio/openmeter/openmeter/testutils"
+	"github.com/openmeterio/openmeter/pkg/pagination"
+)
+
+func Test_Adapter(t *testing.T) {
+	env := NewTestEnv(t)
+	t.Cleanup(func() {
+		env.Close(t)
+	})
+
+	env.DBSchemaMigrate(t)
+
+	namespace := NewTestNamespace(t)
+
+	t.Run("Meter", func(t *testing.T) {
+		t.Run("List", func(t *testing.T) {
+			meterInputs := []meter.CreateMeterInput{
+				{
+					Namespace:     namespace,
+					Name:          "Test meter 1",
+					Key:           "test-meter-1",
+					Aggregation:   meter.MeterAggregationSum,
+					EventType:     "om.meter",
+					ValueProperty: lo.ToPtr("$.value"),
+					GroupBy: map[string]string{
+						"group":   "$.group",
+						"group_2": "$.group_2",
+					},
+				},
+				{
+					Namespace:     namespace,
+					Name:          "Test meter 2",
+					Key:           "test-meter-2",
+					Aggregation:   meter.MeterAggregationCount,
+					EventType:     "om.meter",
+					ValueProperty: nil,
+					GroupBy: map[string]string{
+						"group":   "$.group",
+						"group_2": "$.group_2",
+					},
+				},
+			}
+
+			for _, input := range meterInputs {
+				_, err := env.Meter.CreateMeter(t.Context(), input)
+				require.NoErrorf(t, err, "creating meter must not fail")
+			}
+
+			t.Run("FilterByEventTypes", func(t *testing.T) {
+				out, err := env.Meter.ListMeters(t.Context(), meter.ListMetersParams{
+					Page: pagination.Page{
+						PageSize:   100,
+						PageNumber: 1,
+					},
+					Namespace: namespace,
+					EventTypes: lo.ToPtr([]string{
+						"om.meter",
+					}),
+				})
+				require.NoErrorf(t, err, "listing meters must not fail")
+
+				require.Lenf(t, out.Items, 2, "expected 2 meters with event type om.meter, got %d", len(out.Items))
+
+				for _, m := range out.Items {
+					assert.Equalf(t, m.EventType, "om.meter", "expected meter event type om.meter, got %s", m.EventType)
+				}
+			})
+		})
+	})
+}
+
+type TestEnv struct {
+	Logger *slog.Logger
+	Meter  *Adapter
+
+	Client *entdb.Client
+	db     *testutils.TestDB
+	close  sync.Once
+}
+
+func (e *TestEnv) DBSchemaMigrate(t *testing.T) {
+	t.Helper()
+
+	require.NotNilf(t, e.db, "database must be initialized")
+
+	err := e.db.EntDriver.Client().Schema.Create(t.Context())
+	require.NoErrorf(t, err, "schema migration must not fail")
+}
+
+func (e *TestEnv) Close(t *testing.T) {
+	t.Helper()
+
+	e.close.Do(func() {
+		if e.db != nil {
+			if err := e.db.EntDriver.Close(); err != nil {
+				t.Errorf("failed to close ent driver: %v", err)
+			}
+
+			if err := e.db.PGDriver.Close(); err != nil {
+				t.Errorf("failed to postgres driver: %v", err)
+			}
+		}
+
+		if e.Client != nil {
+			if err := e.Client.Close(); err != nil {
+				t.Errorf("failed to close ent client: %v", err)
+			}
+		}
+	})
+}
+
+func NewTestEnv(t *testing.T) *TestEnv {
+	t.Helper()
+
+	// Init logger
+	logger := testutils.NewDiscardLogger(t)
+
+	// Init database
+	db := testutils.InitPostgresDB(t)
+	client := db.EntDriver.Client()
+
+	// Init meter service
+	meterAdapter, err := New(Config{
+		Client: client,
+		Logger: logger,
+	})
+	require.NoErrorf(t, err, "initializing meter adapter must not fail")
+	require.NotNilf(t, meterAdapter, "meter adapter must not be nil")
+
+	return &TestEnv{
+		Logger: logger,
+		Meter:  meterAdapter,
+		db:     db,
+		Client: client,
+	}
+}
+
+func NewTestULID(t *testing.T) string {
+	t.Helper()
+
+	return ulid.MustNew(ulid.Timestamp(time.Now().UTC()), rand.Reader).String()
+}
+
+var NewTestNamespace = NewTestULID

openmeter/meter/adapter/adapter_test.go

@@ -38,9 +38,14 @@ func (a *Adapter) ListMeters(ctx context.Context, params meter.ListMetersParams)
 		query = query.Where(meterdb.KeyIn(*params.SlugFilter...))
 	}
 
+	if params.EventTypes != nil {
+		query = query.Where(meterdb.EventTypeIn(*params.EventTypes...))
+	}
+
 	// Ordering
 	if params.Order != "" {
-		order := []sql.OrderTermOption{}
+		var order []sql.OrderTermOption
+
 		if !params.Order.IsDefaultValue() {
 			order = entutils.GetOrdering(params.Order)
 		}

openmeter/meter/adapter/meter.go

@@ -73,6 +73,9 @@ type ListMetersParams struct {
 
 	// IncludeDeleted is a flag to include deleted meters in the list.
 	IncludeDeleted bool
+
+	// Filter by event types
+	EventTypes *[]string
 }
 
 // Validate validates the list meters parameters.
@@ -95,7 +98,14 @@ func (p ListMetersParams) Validate() error {
 		for _, slug := range *p.SlugFilter {
 			if slug == "" {
 				errs = append(errs, errors.New("slug filter must not contain empty string"))
-				break
+			}
+		}
+	}
+
+	if p.EventTypes != nil {
+		for _, eventType := range *p.EventTypes {
+			if eventType == "" {
+				errs = append(errs, errors.New("event type filter must not contain empty string"))
 			}
 		}
 	}

openmeter/meter/service.go

@@ -0,0 +1,2 @@
+-- reverse: create index "meter_namespace_event_type" to table: "meters"
+DROP INDEX "meter_namespace_event_type";

tools/migrate/migrations/20251211145809_meter_event_type.down.sql

@@ -0,0 +1,2 @@
+-- create index "meter_namespace_event_type" to table: "meters"
+CREATE INDEX "meter_namespace_event_type" ON "meters" ("namespace", "event_type");

tools/migrate/migrations/20251211145809_meter_event_type.up.sql

@@ -1,4 +1,4 @@
-h1:si+DV0AgNwKY2ymi4F7OArThOL1EUI2xjamCEt/aF7U=
+h1:oV9bvgJUO9Q1fn6Efxv39DURWnVmI6HcrgklVNegg6c=
 20240826120919_init.up.sql h1:tc1V91/smlmaeJGQ8h+MzTEeFjjnrrFDbDAjOYJK91o=
 20240903155435_entitlement-expired-index.up.sql h1:Hp8u5uckmLXc1cRvWU0AtVnnK8ShlpzZNp8pbiJLhac=
 20240917172257_billing-entities.up.sql h1:Q1dAMo0Vjiit76OybClNfYPGC5nmvov2/M2W1ioi4Kw=
@@ -140,3 +140,4 @@ h1:si+DV0AgNwKY2ymi4F7OArThOL1EUI2xjamCEt/aF7U=
 20251121083843_meter_annotations.up.sql h1:9CB03JH2Py9/0yLIrZtKXyqQZxSaxH+4Dys9L/DwXNE=
 20251127184619_invoice-payment-processing-entered-at.up.sql h1:NlpkZurSDozGkRYcCARlNp6nSXbk3jqZeO5YYPEnM20=
 20251210162410_make-customer-usage-attribution-optional.up.sql h1:ZtEKJl8Yj1QS2Ke0lBd8fc7ctHBLPb3XkbJ6peeN1zA=
+20251211145809_meter_event_type.up.sql h1:sG9Y9qUrA+tIjeHdSasBKHJRisBjXgm6nsMQMh1czXg=