fixed notifications dispatch

api/pkg/messaging/internal/notifications/site/notification.go

@@ -10,72 +10,84 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-type DemoRequestNotification struct {
+type SiteRequestNotification struct {
 	messaging.Envelope
-	request *model.DemoRequest
+	requestType    gmessaging.SiteRequestEvent_RequestType
+	demoRequest    *model.DemoRequest
+	contactRequest *model.ContactRequest
 }
 
-func (drn *DemoRequestNotification) Serialize() ([]byte, error) {
-	if drn.request == nil {
-		return nil, merrors.InvalidArgument("demo request payload is empty", "request")
+func (srn *SiteRequestNotification) Serialize() ([]byte, error) {
+	msg := gmessaging.SiteRequestEvent{
+		Type: srn.requestType,
 	}
-	msg := gmessaging.DemoRequestEvent{
-		Name:             drn.request.Name,
-		OrganizationName: drn.request.OrganizationName,
-		Phone:            drn.request.Phone,
-		WorkEmail:        drn.request.WorkEmail,
-		PayoutVolume:     drn.request.PayoutVolume,
-		Comment:          drn.request.Comment,
+
+	switch srn.requestType {
+	case gmessaging.SiteRequestEvent_REQUEST_TYPE_DEMO:
+		if srn.demoRequest == nil {
+			return nil, merrors.InvalidArgument("demo request payload is empty", "request")
+		}
+		msg.Payload = &gmessaging.SiteRequestEvent_Demo{
+			Demo: &gmessaging.SiteDemoRequest{
+				Name:             srn.demoRequest.Name,
+				OrganizationName: srn.demoRequest.OrganizationName,
+				Phone:            srn.demoRequest.Phone,
+				WorkEmail:        srn.demoRequest.WorkEmail,
+				PayoutVolume:     srn.demoRequest.PayoutVolume,
+				Comment:          srn.demoRequest.Comment,
+			},
+		}
+	case gmessaging.SiteRequestEvent_REQUEST_TYPE_CONTACT:
+		if srn.contactRequest == nil {
+			return nil, merrors.InvalidArgument("contact request payload is empty", "request")
+		}
+		msg.Payload = &gmessaging.SiteRequestEvent_Contact{
+			Contact: &gmessaging.SiteContactRequest{
+				Name:    srn.contactRequest.Name,
+				Email:   srn.contactRequest.Email,
+				Phone:   srn.contactRequest.Phone,
+				Company: srn.contactRequest.Company,
+				Topic:   srn.contactRequest.Topic,
+				Message: srn.contactRequest.Message,
+			},
+		}
+	default:
+		return nil, merrors.InvalidArgument("unsupported site request type", "type")
 	}
+
 	data, err := proto.Marshal(&msg)
 	if err != nil {
 		return nil, err
 	}
-	return drn.Envelope.Wrap(data)
+	return srn.Envelope.Wrap(data)
+}
+
+func newSiteRequestEvent() model.NotificationEvent {
+	return model.NewNotification(mservice.Site, nm.NACreated)
 }
 
 func NewDemoRequestEvent() model.NotificationEvent {
-	return model.NewNotification(mservice.Site, nm.NACreated)
-}
-
-func NewDemoRequestEnvelope(sender string, request *model.DemoRequest) messaging.Envelope {
-	return &DemoRequestNotification{
-		Envelope: messaging.CreateEnvelope(sender, NewDemoRequestEvent()),
-		request:  request,
-	}
-}
-
-type ContactRequestNotification struct {
-	messaging.Envelope
-	request *model.ContactRequest
-}
-
-func (crn *ContactRequestNotification) Serialize() ([]byte, error) {
-	if crn.request == nil {
-		return nil, merrors.InvalidArgument("contact request payload is empty", "request")
-	}
-	msg := gmessaging.ContactRequestEvent{
-		Name:    crn.request.Name,
-		Email:   crn.request.Email,
-		Phone:   crn.request.Phone,
-		Company: crn.request.Company,
-		Topic:   crn.request.Topic,
-		Message: crn.request.Message,
-	}
-	data, err := proto.Marshal(&msg)
-	if err != nil {
-		return nil, err
-	}
-	return crn.Envelope.Wrap(data)
+	return newSiteRequestEvent()
 }
 
 func NewContactRequestEvent() model.NotificationEvent {
-	return model.NewNotification(mservice.Site, nm.NACreated)
+	return newSiteRequestEvent()
+}
+
+func NewDemoRequestEnvelope(sender string, request *model.DemoRequest) messaging.Envelope {
+	return &SiteRequestNotification{
+		Envelope:       messaging.CreateEnvelope(sender, newSiteRequestEvent()),
+		requestType:    gmessaging.SiteRequestEvent_REQUEST_TYPE_DEMO,
+		demoRequest:    request,
+		contactRequest: nil,
+	}
 }
 
 func NewContactRequestEnvelope(sender string, request *model.ContactRequest) messaging.Envelope {
-	return &ContactRequestNotification{
-		Envelope: messaging.CreateEnvelope(sender, NewContactRequestEvent()),
-		request:  request,
+	return &SiteRequestNotification{
+		Envelope:       messaging.CreateEnvelope(sender, newSiteRequestEvent()),
+		requestType:    gmessaging.SiteRequestEvent_REQUEST_TYPE_CONTACT,
+		contactRequest: request,
+		demoRequest:    nil,
 	}
 }

api/pkg/messaging/notifications/site/processor.go

@@ -14,72 +14,74 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-type DemoRequestProcessor struct {
-	logger  mlogger.Logger
-	handler handler.DemoRequestHandler
-	event   model.NotificationEvent
+type SiteRequestProcessor struct {
+	logger         mlogger.Logger
+	demoHandler    handler.DemoRequestHandler
+	contactHandler handler.ContactRequestHandler
+	event          model.NotificationEvent
 }
 
-func (drp *DemoRequestProcessor) Process(ctx context.Context, envelope me.Envelope) error {
-	var msg gmessaging.DemoRequestEvent
+func (srp *SiteRequestProcessor) Process(ctx context.Context, envelope me.Envelope) error {
+	var msg gmessaging.SiteRequestEvent
 	if err := proto.Unmarshal(envelope.GetData(), &msg); err != nil {
-		drp.logger.Warn("Failed to decode demo request envelope", zap.Error(err), zap.String("topic", drp.event.ToString()))
+		srp.logger.Warn("Failed to decode site request envelope", zap.Error(err), zap.String("topic", srp.event.ToString()))
 		return err
 	}
-	request := &model.DemoRequest{
-		Name:             msg.GetName(),
-		OrganizationName: msg.GetOrganizationName(),
-		Phone:            msg.GetPhone(),
-		WorkEmail:        msg.GetWorkEmail(),
-		PayoutVolume:     msg.GetPayoutVolume(),
-		Comment:          msg.GetComment(),
-	}
-	return drp.handler(ctx, request)
-}
 
-func (drp *DemoRequestProcessor) GetSubject() model.NotificationEvent {
-	return drp.event
-}
-
-func NewDemoRequestProcessor(logger mlogger.Logger, handler handler.DemoRequestHandler) np.EnvelopeProcessor {
-	return &DemoRequestProcessor{
-		logger:  logger.Named("demo_request_processor"),
-		handler: handler,
-		event:   internalsite.NewDemoRequestEvent(),
+	switch msg.GetType() {
+	case gmessaging.SiteRequestEvent_REQUEST_TYPE_DEMO:
+		if srp.demoHandler == nil {
+			srp.logger.Warn("Demo request handler is not configured")
+			return nil
+		}
+		demo := msg.GetDemo()
+		if demo == nil {
+			srp.logger.Warn("Demo request payload is empty")
+			return nil
+		}
+		request := &model.DemoRequest{
+			Name:             demo.GetName(),
+			OrganizationName: demo.GetOrganizationName(),
+			Phone:            demo.GetPhone(),
+			WorkEmail:        demo.GetWorkEmail(),
+			PayoutVolume:     demo.GetPayoutVolume(),
+			Comment:          demo.GetComment(),
+		}
+		return srp.demoHandler(ctx, request)
+	case gmessaging.SiteRequestEvent_REQUEST_TYPE_CONTACT:
+		if srp.contactHandler == nil {
+			srp.logger.Warn("Contact request handler is not configured")
+			return nil
+		}
+		contact := msg.GetContact()
+		if contact == nil {
+			srp.logger.Warn("Contact request payload is empty")
+			return nil
+		}
+		request := &model.ContactRequest{
+			Name:    contact.GetName(),
+			Email:   contact.GetEmail(),
+			Phone:   contact.GetPhone(),
+			Company: contact.GetCompany(),
+			Topic:   contact.GetTopic(),
+			Message: contact.GetMessage(),
+		}
+		return srp.contactHandler(ctx, request)
+	default:
+		srp.logger.Warn("Received site request with unsupported type", zap.Any("type", msg.GetType()))
+		return nil
 	}
 }
 
-type ContactRequestProcessor struct {
-	logger  mlogger.Logger
-	handler handler.ContactRequestHandler
-	event   model.NotificationEvent
+func (srp *SiteRequestProcessor) GetSubject() model.NotificationEvent {
+	return srp.event
 }
 
-func (crp *ContactRequestProcessor) Process(ctx context.Context, envelope me.Envelope) error {
-	var msg gmessaging.ContactRequestEvent
-	if err := proto.Unmarshal(envelope.GetData(), &msg); err != nil {
-		crp.logger.Warn("Failed to decode contact request envelope", zap.Error(err), zap.String("topic", crp.event.ToString()))
-		return err
-	}
-	request := &model.ContactRequest{
-		Name:    msg.GetName(),
-		Email:   msg.GetEmail(),
-		Phone:   msg.GetPhone(),
-		Company: msg.GetCompany(),
-		Topic:   msg.GetTopic(),
-		Message: msg.GetMessage(),
-	}
-	return crp.handler(ctx, request)
-}
-
-func (crp *ContactRequestProcessor) GetSubject() model.NotificationEvent {
-	return crp.event
-}
-
-func NewContactRequestProcessor(logger mlogger.Logger, handler handler.ContactRequestHandler) np.EnvelopeProcessor {
-	return &ContactRequestProcessor{
-		logger:  logger.Named("contact_request_processor"),
-		handler: handler,
-		event:   internalsite.NewContactRequestEvent(),
+func NewSiteRequestProcessor(logger mlogger.Logger, demo handler.DemoRequestHandler, contact handler.ContactRequestHandler) np.EnvelopeProcessor {
+	return &SiteRequestProcessor{
+		logger:         logger.Named("site_request_processor"),
+		demoHandler:    demo,
+		contactHandler: contact,
+		event:          internalsite.NewDemoRequestEvent(),
 	}
 }

api/pkg/model/notificationevent.go

@@ -71,7 +71,12 @@ func FromString(s string) (*NotificationEventImp, error) {
 
 func StringToNotificationAction(s string) (nm.NotificationAction, error) {
 	switch nm.NotificationAction(s) {
-	case nm.NACreated, nm.NAPending, nm.NAUpdated, nm.NADeleted, nm.NAAssigned, nm.NAPasswordReset:
+	case nm.NACreated,
+		nm.NAPending,
+		nm.NAUpdated,
+		nm.NADeleted,
+		nm.NAAssigned,
+		nm.NAPasswordReset:
 		return nm.NotificationAction(s), nil
 	default:
 		return "", merrors.DataConflict("invalid Notification action: " + s)