upstream-watermill/components/cqrs/event_processor.go

package cqrs

import (
	stdErrors "errors"
	"fmt"

	"github.com/pkg/errors"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
)

type EventProcessorConfig struct {
	// GenerateSubscribeTopic is used to generate topic for subscribing to events.
	// If event processor is using handler groups, GenerateSubscribeTopic is used instead.
	GenerateSubscribeTopic EventProcessorGenerateSubscribeTopicFn

	// SubscriberConstructor is used to create subscriber for EventHandler.
	//
	// This function is called for every EventHandler instance.
	// If you want to re-use one subscriber for multiple handlers, use GroupEventProcessor instead.
	SubscriberConstructor EventProcessorSubscriberConstructorFn

	// OnHandle is called before handling event.
	// OnHandle works in a similar way to middlewares: you can inject additional logic before and after handling an event.
	//
	// Because of that, you need to explicitly call params.Handler.Handle() to handle the event.
	//
	//  func(params EventProcessorOnHandleParams) (err error) {
	//      // logic before handle
	//      //  (...)
	//
	//      err := params.Handler.Handle(params.Message.Context(), params.Event)
	//
	//      // logic after handle
	//      //  (...)
	//
	//      return err
	//  }
	//
	// This option is not required.
	OnHandle EventProcessorOnHandleFn

	// AckOnUnknownEvent is used to decide if message should be acked if event has no handler defined.
	AckOnUnknownEvent bool

	// Marshaler is used to marshal and unmarshal events.
	// It is required.
	Marshaler CommandEventMarshaler

	// Logger instance used to log.
	// If not provided, watermill.NopLogger is used.
	Logger watermill.LoggerAdapter

	// disableRouterAutoAddHandlers is used to keep backwards compatibility.
	// it is set when EventProcessor is created by NewEventProcessor.
	// Deprecated: please migrate to NewEventProcessorWithConfig.
	disableRouterAutoAddHandlers bool
}

func (c *EventProcessorConfig) setDefaults() {
	if c.Logger == nil {
		c.Logger = watermill.NopLogger{}
	}
}

func (c EventProcessorConfig) Validate() error {
	var err error

	if c.Marshaler == nil {
		err = stdErrors.Join(err, errors.New("missing Marshaler"))
	}

	if c.GenerateSubscribeTopic == nil {
		err = stdErrors.Join(err, errors.New("missing GenerateHandlerTopic"))
	}
	if c.SubscriberConstructor == nil {
		err = stdErrors.Join(err, errors.New("missing SubscriberConstructor"))
	}

	return err
}

type EventProcessorGenerateSubscribeTopicFn func(EventProcessorGenerateSubscribeTopicParams) (string, error)

type EventProcessorGenerateSubscribeTopicParams struct {
	EventName    string
	EventHandler EventHandler
}

type EventProcessorSubscriberConstructorFn func(EventProcessorSubscriberConstructorParams) (message.Subscriber, error)

type EventProcessorSubscriberConstructorParams struct {
	EventName    string
	HandlerName  string
	EventHandler EventHandler
}

type EventProcessorOnHandleFn func(params EventProcessorOnHandleParams) error

type EventProcessorOnHandleParams struct {
	Handler EventHandler

	Event     any
	EventName string

	// Message is never nil and can be modified.
	Message *message.Message
}

// EventProcessor determines which EventHandler should handle event received from event bus.
type EventProcessor struct {
	router   *message.Router
	handlers []EventHandler
	config   EventProcessorConfig
}

// NewEventProcessorWithConfig creates a new EventProcessor.
func NewEventProcessorWithConfig(router *message.Router, config EventProcessorConfig) (*EventProcessor, error) {
	config.setDefaults()

	if err := config.Validate(); err != nil {
		return nil, errors.Wrap(err, "invalid config EventProcessor")
	}
	if router == nil && !config.disableRouterAutoAddHandlers {
		return nil, errors.New("missing router")
	}

	return &EventProcessor{
		router: router,
		config: config,
	}, nil
}

// NewEventProcessor creates a new EventProcessor.
// Deprecated. Use NewEventProcessorWithConfig instead.
func NewEventProcessor(
	individualHandlers []EventHandler,
	generateTopic func(eventName string) string,
	subscriberConstructor EventsSubscriberConstructor,
	marshaler CommandEventMarshaler,
	logger watermill.LoggerAdapter,
) (*EventProcessor, error) {
	if len(individualHandlers) == 0 {
		return nil, errors.New("missing handlers")
	}
	if generateTopic == nil {
		return nil, errors.New("nil generateTopic")
	}
	if subscriberConstructor == nil {
		return nil, errors.New("missing subscriberConstructor")
	}
	if marshaler == nil {
		return nil, errors.New("missing marshaler")
	}
	if logger == nil {
		logger = watermill.NopLogger{}
	}

	eventProcessorConfig := EventProcessorConfig{
		AckOnUnknownEvent: true, // this is the previous default behaviour - keeping backwards compatibility
		GenerateSubscribeTopic: func(params EventProcessorGenerateSubscribeTopicParams) (string, error) {
			return generateTopic(params.EventName), nil
		},
		SubscriberConstructor: func(params EventProcessorSubscriberConstructorParams) (message.Subscriber, error) {
			return subscriberConstructor(params.HandlerName)
		},
		Marshaler:                    marshaler,
		Logger:                       logger,
		disableRouterAutoAddHandlers: true,
	}
	eventProcessorConfig.setDefaults()

	ep, err := NewEventProcessorWithConfig(nil, eventProcessorConfig)
	if err != nil {
		return nil, err
	}

	for _, handler := range individualHandlers {
		if err := ep.AddHandlers(handler); err != nil {
			return nil, err
		}
	}

	return ep, nil
}

// EventsSubscriberConstructor creates a subscriber for EventHandler.
// It allows you to create separated customized Subscriber for every command handler.
//
// When handler groups are used, handler group is passed as handlerName.
// Deprecated: please use EventProcessorSubscriberConstructorFn instead.
type EventsSubscriberConstructor func(handlerName string) (message.Subscriber, error)

// AddHandlers adds a new EventHandler to the EventProcessor and adds it to the router.
func (p *EventProcessor) AddHandlers(handlers ...EventHandler) error {
	if p.config.disableRouterAutoAddHandlers {
		p.handlers = append(p.handlers, handlers...)
		return nil
	}

	for _, handler := range handlers {
		if _, err := p.addHandlerToRouter(p.router, handler); err != nil {
			return err
		}

		p.handlers = append(p.handlers, handler)
	}

	return nil
}

// AddHandler adds a new EventHandler to the EventProcessor and adds it to the router.
func (p *EventProcessor) AddHandler(handler EventHandler) (*message.Handler, error) {
	if p.config.disableRouterAutoAddHandlers {
		p.handlers = append(p.handlers, handler)

		return nil, nil
	}

	h, err := p.addHandlerToRouter(p.router, handler)
	if err != nil {
		return nil, err
	}

	p.handlers = append(p.handlers, handler)

	return h, nil
}

// AddHandlersToRouter adds the EventProcessor's handlers to the given router.
// It should be called only once per EventProcessor instance.
//
// It is required to call AddHandlersToRouter only if command processor is created with NewEventProcessor (disableRouterAutoAddHandlers is set to true).
// Deprecated: please migrate to event processor created by NewEventProcessorWithConfig.
func (p EventProcessor) AddHandlersToRouter(r *message.Router) error {
	if !p.config.disableRouterAutoAddHandlers {
		return errors.New("AddHandlersToRouter should be called only when using deprecated NewEventProcessor")
	}

	for i := range p.handlers {
		handler := p.handlers[i]

		if _, err := p.addHandlerToRouter(r, handler); err != nil {
			return err
		}
	}

	return nil
}

func (p EventProcessor) addHandlerToRouter(r *message.Router, handler EventHandler) (*message.Handler, error) {
	if err := validateEvent(handler.NewEvent()); err != nil {
		return nil, errors.Wrapf(err, "invalid event for handler %s", handler.HandlerName())
	}

	handlerName := handler.HandlerName()
	eventName := p.config.Marshaler.Name(handler.NewEvent())

	topicName, err := p.config.GenerateSubscribeTopic(EventProcessorGenerateSubscribeTopicParams{
		EventName:    eventName,
		EventHandler: handler,
	})
	if err != nil {
		return nil, errors.Wrapf(err, "cannot generate topic name for handler %s", handlerName)
	}

	logger := p.config.Logger.With(watermill.LogFields{
		"event_handler_name": handlerName,
		"topic":              topicName,
	})

	handlerFunc, err := p.routerHandlerFunc(handler, logger)
	if err != nil {
		return nil, err
	}

	if p.config.SubscriberConstructor == nil {
		return nil, errors.New("missing SubscriberConstructor config option")
	}

	subscriber, err := p.config.SubscriberConstructor(EventProcessorSubscriberConstructorParams{
		EventName:    eventName,
		HandlerName:  handlerName,
		EventHandler: handler,
	})
	if err != nil {
		return nil, errors.Wrap(err, "cannot create subscriber for event processor")
	}

	return addHandlerToRouter(p.config.Logger, r, handlerName, topicName, handlerFunc, subscriber), nil
}

func (p EventProcessor) Handlers() []EventHandler {
	return p.handlers
}

func addHandlerToRouter(logger watermill.LoggerAdapter, r *message.Router, handlerName string, topicName string, handlerFunc message.NoPublishHandlerFunc, subscriber message.Subscriber) *message.Handler {
	logger = logger.With(watermill.LogFields{
		"event_handler_name": handlerName,
		"topic":              topicName,
	})

	logger.Debug("Adding CQRS event handler to router", nil)

	return r.AddConsumerHandler(
		handlerName,
		topicName,
		subscriber,
		handlerFunc,
	)
}

func (p EventProcessor) routerHandlerFunc(handler EventHandler, logger watermill.LoggerAdapter) (message.NoPublishHandlerFunc, error) {
	initEvent := handler.NewEvent()
	expectedEventName := p.config.Marshaler.Name(initEvent)

	if err := validateEvent(initEvent); err != nil {
		return nil, err
	}

	return func(msg *message.Message) error {
		event := handler.NewEvent()
		messageEventName := p.config.Marshaler.NameFromMessage(msg)

		if messageEventName != expectedEventName {
			if !p.config.AckOnUnknownEvent {
				return fmt.Errorf("received unexpected event type %s, expected %s", messageEventName, expectedEventName)
			} else {
				logger.Trace("Received different event type than expected, ignoring", watermill.LogFields{
					"message_uuid":        msg.UUID,
					"expected_event_type": expectedEventName,
					"received_event_type": messageEventName,
				})
				return nil
			}
		}

		logger.Debug("Handling event", watermill.LogFields{
			"message_uuid":        msg.UUID,
			"received_event_type": messageEventName,
		})

		ctx := CtxWithOriginalMessage(msg.Context(), msg)
		msg.SetContext(ctx)

		if err := p.config.Marshaler.Unmarshal(msg, event); err != nil {
			return err
		}

		handle := func(params EventProcessorOnHandleParams) error {
			return params.Handler.Handle(ctx, params.Event)
		}
		if p.config.OnHandle != nil {
			handle = p.config.OnHandle
		}

		err := handle(EventProcessorOnHandleParams{
			Handler:   handler,
			Event:     event,
			EventName: messageEventName,
			Message:   msg,
		})
		if err != nil {
			logger.Debug("Error when handling event", watermill.LogFields{"err": err})
			return err
		}

		return nil
	}, nil
}

func validateEvent(event interface{}) error {
	// EventHandler's NewEvent must return a pointer, because it is used to unmarshal
	if err := isPointer(event); err != nil {
		return errors.Wrap(err, "command must be a non-nil pointer")
	}

	return nil
}