Merge pull request #1602 from openmeterio/fix-svix-timeout

refactor: mk webhook event type register non-critical
openmeterio · Oct 1, 2024 · e63cd3b · e63cd3b
2 parents 0efae86 + db06a75
commit e63cd3b
Show file tree

Hide file tree

Showing 9 changed files with 66 additions and 17 deletions.
diff --git a/cmd/notification-service/main.go b/cmd/notification-service/main.go
@@ -42,7 +42,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/watermill/eventbus"
 	"github.com/openmeterio/openmeter/openmeter/watermill/router"
 	"github.com/openmeterio/openmeter/pkg/contextx"
-	entdriver "github.com/openmeterio/openmeter/pkg/framework/entutils/entdriver"
+	"github.com/openmeterio/openmeter/pkg/framework/entutils/entdriver"
 	"github.com/openmeterio/openmeter/pkg/framework/operation"
 	"github.com/openmeterio/openmeter/pkg/framework/pgdriver"
 	"github.com/openmeterio/openmeter/pkg/gosundheit"
@@ -305,7 +305,10 @@ func main() {
 	}
 
 	notificationWebhook, err := notificationwebhook.New(notificationwebhook.Config{
-		SvixConfig: conf.Svix,
+		SvixConfig:              conf.Svix,
+		RegistrationTimeout:     conf.Notification.Webhook.EventTypeRegistrationTimeout,
+		SkipRegistrationOnError: conf.Notification.Webhook.SkipEventTypeRegistrationOnError,
+		Logger:                  logger.WithGroup("notification.webhook"),
 	})
 	if err != nil {
 		logger.Error("failed to initialize notification repository", "error", err)

diff --git a/cmd/server/main.go b/cmd/server/main.go
@@ -422,7 +422,10 @@ func main() {
 
 		var notificationWebhook notificationwebhook.Handler
 		notificationWebhook, err = notificationwebhook.New(notificationwebhook.Config{
-			SvixConfig: conf.Svix,
+			SvixConfig:              conf.Svix,
+			RegistrationTimeout:     conf.Notification.Webhook.EventTypeRegistrationTimeout,
+			SkipRegistrationOnError: conf.Notification.Webhook.SkipEventTypeRegistrationOnError,
+			Logger:                  logger.WithGroup("notification.webhook"),
 		})
 		if err != nil {
 			logger.Error("failed to initialize notification webhook handler", "error", err)

diff --git a/config.example.yaml b/config.example.yaml
@@ -92,6 +92,9 @@ meters:
 
 notification:
   enabled: true
+#  webhook:
+#    eventTypeRegistrationTimeout: 30s
+#    skipEventTypeRegistrationOnError: false
 
 svix:
   apiKey: secret

diff --git a/config/config_test.go b/config/config_test.go
@@ -273,6 +273,10 @@ func TestComplete(t *testing.T) {
 				},
 				ConsumerGroupName: "om_notification_service",
 			},
+			Webhook: WebhookConfiguration{
+				EventTypeRegistrationTimeout:     notificationwebhook.DefaultRegistrationTimeout,
+				SkipEventTypeRegistrationOnError: false,
+			},
 		},
 		Svix: notificationwebhook.SvixConfig{
 			APIKey:    "test-svix-token",

diff --git a/config/notification.go b/config/notification.go
@@ -2,13 +2,25 @@ package config
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/spf13/viper"
+
+	"github.com/openmeterio/openmeter/openmeter/notification/webhook"
 )
 
+type WebhookConfiguration struct {
+	// Timeout for registering event types in webhook provider
+	EventTypeRegistrationTimeout time.Duration
+	// Skip registering event types on unsuccessful attempt instead of returning with error
+	SkipEventTypeRegistrationOnError bool
+}
+
 type NotificationConfiguration struct {
 	Enabled  bool
 	Consumer ConsumerConfiguration
+
+	Webhook WebhookConfiguration
 }
 
 func (c NotificationConfiguration) Validate() error {
@@ -22,4 +34,6 @@ func ConfigureNotification(v *viper.Viper) {
 	ConfigureConsumer(v, "notification.consumer")
 	v.SetDefault("notification.consumer.dlq.topic", "om_sys.notification_service_dlq")
 	v.SetDefault("notification.consumer.consumerGroupName", "om_notification_service")
+	v.SetDefault("notification.webhook.eventTypeRegistrationTimeout", webhook.DefaultRegistrationTimeout)
+	v.SetDefault("notification.webhook.skipEventTypeRegistrationOnError", false)
 }
diff --git a/openmeter/notification/webhook/svix.go b/openmeter/notification/webhook/svix.go
@@ -76,15 +76,15 @@ func newSvixWebhookHandler(config SvixConfig) (Handler, error) {
 }
 
 func (h svixWebhookHandler) RegisterEventTypes(ctx context.Context, params RegisterEventTypesInputs) error {
-	for _, evenType := range params.EvenTypes {
+	for _, eventType := range params.EventTypes {
 		input := &svix.EventTypeUpdate{
-			Description: evenType.Description,
+			Description: eventType.Description,
 			FeatureFlag: *svix.NullableString(nil),
-			GroupName:   *svix.NullableString(&evenType.GroupName),
-			Schemas:     evenType.Schemas,
+			GroupName:   *svix.NullableString(&eventType.GroupName),
+			Schemas:     eventType.Schemas,
 		}
 
-		_, err := h.client.EventType.Update(ctx, evenType.Name, input)
+		_, err := h.client.EventType.Update(ctx, eventType.Name, input)
 		if err != nil {
 			err = unwrapSvixError(err)
 

diff --git a/openmeter/notification/webhook/webhook.go b/openmeter/notification/webhook/webhook.go
@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"log/slog"
 	"strings"
 	"time"
 )
@@ -240,7 +241,7 @@ func (i SendMessageInput) Validate() error {
 }
 
 type RegisterEventTypesInputs struct {
-	EvenTypes   []EventType
+	EventTypes  []EventType
 	AllowUpdate bool
 }
 
@@ -263,31 +264,51 @@ type Handler interface {
 	SendMessage(ctx context.Context, params SendMessageInput) (*Message, error)
 }
 
+const (
+	DefaultRegistrationTimeout = 30 * time.Second
+)
+
 type Config struct {
 	SvixConfig
 
-	RegisterEvenTypes []EventType
+	RegisterEventTypes      []EventType
+	RegistrationTimeout     time.Duration
+	SkipRegistrationOnError bool
+
+	Logger *slog.Logger
 }
 
 func New(config Config) (Handler, error) {
-	if config.RegisterEvenTypes == nil {
-		config.RegisterEvenTypes = NotificationEventTypes
+	if config.Logger == nil {
+		return nil, errors.New("logger is required")
+	}
+
+	if config.RegisterEventTypes == nil {
+		config.RegisterEventTypes = NotificationEventTypes
+	}
+
+	if config.RegistrationTimeout == 0 {
+		config.RegistrationTimeout = DefaultRegistrationTimeout
 	}
 
 	handler, err := newSvixWebhookHandler(config.SvixConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Svix webhook handler: %w", err)
 	}
 
-	if len(config.RegisterEvenTypes) > 0 {
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	if len(config.RegisterEventTypes) > 0 {
+		ctx, cancel := context.WithTimeout(context.Background(), config.RegistrationTimeout)
 		defer cancel()
 
 		err = handler.RegisterEventTypes(ctx, RegisterEventTypesInputs{
-			EvenTypes: config.RegisterEvenTypes,
+			EventTypes: config.RegisterEventTypes,
 		})
 		if err != nil {
-			return nil, fmt.Errorf("failed to register event types: %w", err)
+			if config.SkipRegistrationOnError {
+				config.Logger.Warn("failed to register event types", "error", err)
+			} else {
+				return nil, fmt.Errorf("failed to register event types: %w", err)
+			}
 		}
 	}
 

diff --git a/test/notification/testenv.go b/test/notification/testenv.go
@@ -135,6 +135,7 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) {
 			ServerURL: fmt.Sprintf(SvixServerURLTemplate, svixHost),
 			Debug:     false,
 		},
+		Logger: logger,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create webhook handler: %w", err)

diff --git a/test/notification/webhook.go b/test/notification/webhook.go
@@ -41,7 +41,7 @@ type WebhookTestSuite struct {
 
 func (s *WebhookTestSuite) Setup(ctx context.Context, t *testing.T) {
 	err := s.Env.NotificationWebhook().RegisterEventTypes(ctx, notificationwebhook.RegisterEventTypesInputs{
-		EvenTypes: notificationwebhook.NotificationEventTypes,
+		EventTypes: notificationwebhook.NotificationEventTypes,
 	})
 	assert.NoError(t, err, "Registering event types must not fail")
 }