app/common/notification.go

package common

import (
	"fmt"
	"log/slog"
	"math/rand/v2"

	"github.com/google/wire"
	svix "github.com/svix/svix-webhooks/go"
	"go.opentelemetry.io/otel/trace"

	"github.com/openmeterio/openmeter/app/config"
	entdb "github.com/openmeterio/openmeter/openmeter/ent/db"
	"github.com/openmeterio/openmeter/openmeter/notification"
	notificationadapter "github.com/openmeterio/openmeter/openmeter/notification/adapter"
	"github.com/openmeterio/openmeter/openmeter/notification/eventhandler"
	notificationservice "github.com/openmeterio/openmeter/openmeter/notification/service"
	notificationwebhook "github.com/openmeterio/openmeter/openmeter/notification/webhook"
	webhooknoop "github.com/openmeterio/openmeter/openmeter/notification/webhook/noop"
	webhooksvix "github.com/openmeterio/openmeter/openmeter/notification/webhook/svix"
	"github.com/openmeterio/openmeter/openmeter/productcatalog/feature"
	"github.com/openmeterio/openmeter/pkg/framework/pgdriver"
	"github.com/openmeterio/openmeter/pkg/pglockx"
)

var Notification = wire.NewSet(
	NewNotificationAdapter,
	NewNotificationService,
	NewNotificationWebhookHandler,
	NewNotificationEventHandler,
)

// NotificationService is a wire set for the notification service, it can be used at
// places where only the service is required without svix and event handler.
var NotificationService = wire.NewSet(
	NewNotificationAdapter,
	NewNotificationService,
	NewNoopNotificationWebhookHandler,
)

func NewNotificationAdapter(
	logger *slog.Logger,
	db *entdb.Client,
) (notification.Repository, error) {
	adapter, err := notificationadapter.New(notificationadapter.Config{
		Client: db,
		Logger: logger,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to initialize notification adapter: %w", err)
	}

	return adapter, nil
}

func NewNotificationEventHandler(
	config config.NotificationConfiguration,
	logger *slog.Logger,
	tracer trace.Tracer,
	adapter notification.Repository,
	webhook notificationwebhook.Handler,
	driver *pgdriver.Driver,
) (notification.EventHandler, error) {
	config.Lock.Owner = fmt.Sprintf("notification.event_handler-%v", rand.Int())
	config.Lock.HeartbeatInterval = max(pglockx.DefaultHeartbeatInterval, config.Lock.HeartbeatInterval)
	config.Lock.LeaseTime = max(config.Lock.HeartbeatInterval*2, config.Lock.LeaseTime)

	logger.Debug("initializing notification lock client",
		"lock.leaseTime", config.Lock.LeaseTime, "lock.heartbeatInterval", config.Lock.HeartbeatInterval, "lock.owner", config.Lock.Owner)

	lockClient, err := pglockx.New(driver.DB(), config.Lock)
	if err != nil {
		return nil, fmt.Errorf("failed to initialize notification lock client: %w", err)
	}

	eventHandler, err := eventhandler.New(eventhandler.Config{
		Repository:        adapter,
		Webhook:           webhook,
		Logger:            logger,
		Tracer:            tracer,
		ReconcileInterval: config.ReconcileInterval,
		SendingTimeout:    config.SendingTimeout,
		PendingTimeout:    config.PendingTimeout,
		ReconcilerWorkers: config.ReconcilerWorkers,
		LockClient:        lockClient,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to initialize notification event handler: %w", err)
	}

	return eventHandler, nil
}

func NewNotificationService(
	logger *slog.Logger,
	adapter notification.Repository,
	webhook notificationwebhook.Handler,
	featureConnector feature.FeatureConnector,
) (notification.Service, error) {
	notificationService, err := notificationservice.New(notificationservice.Config{
		Adapter:          adapter,
		Webhook:          webhook,
		FeatureConnector: featureConnector,
		Logger:           logger.With(slog.String("subsystem", "notification")),
	})
	if err != nil {
		return nil, fmt.Errorf("failed to initialize notification service: %w", err)
	}

	return notificationService, nil
}

func NewNoopNotificationWebhookHandler(
	logger *slog.Logger,
) (notificationwebhook.Handler, error) {
	return webhooknoop.New(logger), nil
}

func NewNotificationWebhookHandler(
	logger *slog.Logger,
	tracer trace.Tracer,
	webhookConfig config.WebhookConfiguration,
	svixClient *svix.Svix,
) (notificationwebhook.Handler, error) {
	if svixClient == nil {
		logger.Warn("svix client not configured, using noop handler")

		return webhooknoop.New(logger), nil
	}

	handler, err := webhooksvix.New(webhooksvix.Config{
		SvixAPIClient:           svixClient,
		RegisterEventTypes:      notificationwebhook.NotificationEventTypes,
		RegistrationTimeout:     webhookConfig.EventTypeRegistrationTimeout,
		SkipRegistrationOnError: webhookConfig.SkipEventTypeRegistrationOnError,
		Logger:                  logger.WithGroup("notification.webhook"),
		Tracer:                  tracer,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to initialize notification webhook handler: %w", err)
	}

	return handler, nil
}