Merge pull request #1258 from openmeterio/chore/expose-events

chore: expose event subsystem for the cloud deployment

cmd/server/main.go

@@ -499,6 +499,8 @@ func initEventPublisher(ctx context.Context, logger *slog.Logger, conf config.Co
 			return nil, fmt.Errorf("failed to create Kafka admin client: %w", err)
 		}
 
+		defer adminClient.Close()
+
 		if err := pkgkafka.ProvisionTopic(ctx, adminClient, logger, conf.Events.SystemEvents.Topic, conf.Events.SystemEvents.AutoProvision.Partitions); err != nil {
 			return nil, fmt.Errorf("failed to auto provision topic: %w", err)
 		}

internal/watermill/driver/kafka/publisher.go

@@ -1,4 +1,4 @@
-package noop
+package kafka
 
 import (
 	"github.com/ThreeDotsLabs/watermill/message"

openmeter/credit/events.go

@@ -0,0 +1,17 @@
+package credit
+
+import "github.com/openmeterio/openmeter/internal/credit"
+
+const (
+	EventSubsystem = credit.EventSubsystem
+)
+
+const (
+	EventCreateGrant = credit.EventCreateGrant
+	EventVoidGrant   = credit.EventVoidGrant
+)
+
+type (
+	GrantCreatedEvent = credit.GrantCreatedEvent
+	GrantVoidedEvent  = credit.GrantVoidedEvent
+)

openmeter/entitlement/events.go

@@ -0,0 +1,17 @@
+package entitlement
+
+import "github.com/openmeterio/openmeter/internal/entitlement"
+
+const (
+	EventSubsystem = entitlement.EventSubsystem
+)
+
+const (
+	EventCreateEntitlement = entitlement.EventCreateEntitlement
+	EventDeleteEntitlement = entitlement.EventDeleteEntitlement
+)
+
+type (
+	EntitlementCreatedEvent = entitlement.EntitlementCreatedEvent
+	EntitlementDeletedEvent = entitlement.EntitlementDeletedEvent
+)

openmeter/entitlement/metered/events.go

@@ -0,0 +1,15 @@
+package meteredentitlement
+
+import meteredentitlement "github.com/openmeterio/openmeter/internal/entitlement/metered"
+
+const (
+	EventSubsystem = meteredentitlement.EventSubsystem
+)
+
+const (
+	EventResetEntitlementUsage = meteredentitlement.EventResetEntitlementUsage
+)
+
+type (
+	ResetEntitlementEvent = meteredentitlement.ResetEntitlementEvent
+)

openmeter/event/publisher/publisher.go

@@ -7,3 +7,7 @@ type (
 	PublisherOptions = publisher.PublisherOptions
 	TopicPublisher   = publisher.TopicPublisher
 )
+
+func NewPublisher(options PublisherOptions) (Publisher, error) {
+	return publisher.NewPublisher(options)
+}

openmeter/watermill/driver/kafka/driver.go

@@ -0,0 +1,25 @@
+package kafka
+
+import (
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/cloudevents/sdk-go/v2/event"
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+
+	watermillkafka "github.com/openmeterio/openmeter/internal/watermill/driver/kafka"
+)
+
+const (
+	PartitionKeyMetadataKey = watermillkafka.PartitionKeyMetadataKey
+)
+
+type (
+	Publisher = watermillkafka.Publisher
+)
+
+func NewPublisher(producer *kafka.Producer) *Publisher {
+	return watermillkafka.NewPublisher(producer)
+}
+
+func AddPartitionKeyFromSubject(watermillIn *message.Message, cloudEvent event.Event) (*message.Message, error) {
+	return watermillkafka.AddPartitionKeyFromSubject(watermillIn, cloudEvent)
+}

openmeter/watermill/driver/noop/driver.go

@@ -0,0 +1,7 @@
+package noop
+
+import "github.com/openmeterio/openmeter/internal/watermill/driver/noop"
+
+type (
+	Publisher = noop.Publisher
+)