Add ability to consume events

Author: AchoArnold

api/pkg/entities/message.go

@@ -37,6 +37,14 @@ const (
 	MessageStatusFailed = "failed"
 )
 
+// MessageEventName is the type of event generated by the mobile phone for a message
+type MessageEventName string
+
+const (
+	// MessageEventNameSent is emitted when a message is sent by the mobile phone
+	MessageEventNameSent = "SENT"
+)
+
 // Message represents a message sent between 2 phone numbers
 type Message struct {
 	ID      uuid.UUID     `json:"id" gorm:"primaryKey;type:uuid;" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`

api/pkg/events/message_phone_sent_event.go

@@ -1,4 +1,14 @@
 package events
 
+import "github.com/google/uuid"
+
 // EventTypeMessagePhoneSent is emitted when the phone sends a message
 const EventTypeMessagePhoneSent = "message.phone.sent"
+
+// MessagePhoneSentPayload is the payload of the EventTypeMessagePhoneSent event
+type MessagePhoneSentPayload struct {
+	ID      uuid.UUID `json:"id"`
+	From    string    `json:"from"`
+	To      string    `json:"to"`
+	Content string    `json:"content"`
+}

api/pkg/handlers/message_handler.go

@@ -41,6 +41,7 @@ func (h *MessageHandler) RegisterRoutes(router fiber.Router) {
 	router.Post("/messages/send", h.PostSend)
 	router.Get("/messages/outstanding", h.GetOutstanding)
 	router.Get("/messages", h.Index)
+	router.Post("/messages/:messageID/event", h.PostEvent)
 }
 
 // PostSend a new entities.Message
@@ -169,3 +170,46 @@ func (h *MessageHandler) Index(c *fiber.Ctx) error {
 
 	return h.responseOK(c, fmt.Sprintf("fetched %d %s", len(*messages), h.pluralize("message", len(*messages))), messages)
 }
+
+// PostEvent registers an event on a message
+// @Summary      Store an event for a message on the mobile phone
+// @Description  Use this endpoint to send events for a message when it is failed, sent or delivered by the mobile phone.
+// @Tags         Messages
+// @Accept       json
+// @Produce      json
+// @Param 		 messageID 	path		string 							true 	"ID of the message" 			default(32343a19-da5e-4b1b-a767-3298a73703ca)
+// @Param        payload   	body 		requests.MessageEvent  			true 	"Payload of the event emitted."
+// @Success      200  		{object} 	responses.MessageResponse
+// @Success      400  		{object}  	responses.BadRequest
+// @Success      422  		{object} 	responses.UnprocessableEntity
+// @Success      500  		{object}  	responses.InternalServerError
+// @Router       /messages/:messageID/event [post]
+func (h *MessageHandler) PostEvent(c *fiber.Ctx) error {
+	ctx, span := h.tracer.StartFromFiberCtx(c)
+	defer span.End()
+
+	ctxLogger := h.tracer.CtxLogger(h.logger, span)
+
+	var request requests.MessageEvent
+	if err := c.BodyParser(&request); err != nil {
+		msg := fmt.Sprintf("cannot marshall [%s] into %T", c.Body(), request)
+		ctxLogger.Warn(stacktrace.Propagate(err, msg))
+		return h.responseBadRequest(c, err)
+	}
+
+	request.MessageID = c.Params("messageID")
+	if errors := h.validator.ValidateMessageEvent(ctx, request); len(errors) != 0 {
+		msg := fmt.Sprintf("validation errors [%s], while storing event [%s] for message [%s]", spew.Sdump(errors), c.Body(), request.MessageID)
+		ctxLogger.Warn(stacktrace.NewError(msg))
+		return h.responseUnprocessableEntity(c, errors, "validation errors while storing event")
+	}
+
+	message, err := h.service.StoreEvent(ctx, request.ToMessageStoreEventParams(c.OriginalURL()))
+	if err != nil {
+		msg := fmt.Sprintf("cannot store event for message [%s] with paylod [%s]", request.MessageID, c.Body())
+		ctxLogger.Error(stacktrace.Propagate(err, msg))
+		return h.responseInternalServerError(c)
+	}
+
+	return h.responseOK(c, "message event stored successfully", message)
+}

api/pkg/requests/message_sent_request.go

@@ -0,0 +1,34 @@
+package requests
+
+import (
+	"time"
+
+	"github.com/NdoleStudio/http-sms-manager/pkg/entities"
+	"github.com/google/uuid"
+
+	"github.com/NdoleStudio/http-sms-manager/pkg/services"
+)
+
+// MessageEvent is the payload for sending and SMS message
+type MessageEvent struct {
+	// Timestamp is the time when the event was emitted, Please send the timestamp in UTC with as much precision as possible
+	Timestamp time.Time `json:"sent_at" example:"2022-06-05T14:26:09.527976+03:00"`
+
+	// EventName is the type of event
+	// * SENT: is emitted when a message is sent by the mobile phone (only SENT is implemented)
+	// * FAILED: is event is emitted when the message could not be sent by the mobile phone
+	// * DELIVERED: is event is emitted when a delivery report has been received by the mobile phone
+	EventName string `json:"event_name" example:"SENT"`
+
+	MessageID string `json:"messageID" swaggerignore:"true"` // used internally for validation
+}
+
+// ToMessageStoreEventParams converts MessageEvent to services.MessageStorePhoneEventParams
+func (input MessageEvent) ToMessageStoreEventParams(source string) services.MessageStorePhoneEventParams {
+	return services.MessageStorePhoneEventParams{
+		MessageID: uuid.MustParse(input.MessageID),
+		Source:    source,
+		EventName: entities.MessageEventName(input.EventName),
+		Timestamp: input.Timestamp,
+	}
+}

api/pkg/services/message_service.go

@@ -86,6 +86,62 @@ func (service *MessageService) GetMessages(ctx context.Context, params MessageGe
 	return messages, nil
 }
 
+// MessageStorePhoneEventParams parameters registering a message event
+type MessageStorePhoneEventParams struct {
+	MessageID uuid.UUID
+	EventName entities.MessageEventName
+	Timestamp time.Time
+	Source    string
+}
+
+// StoreEvent handles event generated by a mobile phone
+func (service *MessageService) StoreEvent(ctx context.Context, params MessageStorePhoneEventParams) (*entities.Message, error) {
+	ctx, span := service.tracer.Start(ctx)
+	defer span.End()
+
+	message, err := service.repository.Load(ctx, params.MessageID)
+	if err != nil {
+		msg := fmt.Sprintf("could not fetch messages with parms [%+#v]", params)
+		return nil, service.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
+	}
+
+	switch params.EventName {
+	case entities.MessageEventNameSent:
+		err = service.handleMessageSentEvent(ctx, params, message)
+	default:
+		return nil, service.tracer.WrapErrorSpan(span, stacktrace.NewError(fmt.Sprintf("cannot handle message event [%s]", params.EventName)))
+	}
+
+	if err != nil {
+		msg := fmt.Sprintf("could not handle phone event [%s] for message with id [%s]", params.EventName, message.ID)
+		return nil, service.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
+	}
+
+	return service.repository.Load(ctx, params.MessageID)
+}
+
+func (service *MessageService) handleMessageSentEvent(ctx context.Context, params MessageStorePhoneEventParams, message *entities.Message) error {
+	ctx, span := service.tracer.Start(ctx)
+	defer span.End()
+
+	event, err := service.createMessagePhoneSentEvent(params.Source, events.MessagePhoneSentPayload{
+		ID:      message.ID,
+		From:    message.From,
+		To:      message.To,
+		Content: message.Content,
+	})
+	if err != nil {
+		msg := fmt.Sprintf("cannot create event [%s] for message [%s]", events.EventTypeMessagePhoneSent, message.ID)
+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
+	}
+
+	if err = service.eventDispatcher.Dispatch(ctx, event); err != nil {
+		msg := fmt.Sprintf("cannot dispatch event type [%s] and id [%s]", event.Type(), event.ID())
+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
+	}
+	return nil
+}
+
 func (service *MessageService) handleOutstandingMessages(ctx context.Context, source string, messages *[]entities.Message) *[]entities.Message {
 	ctx, span := service.tracer.Start(ctx)
 	defer span.End()
@@ -278,6 +334,10 @@ func (service *MessageService) createMessagePhoneSendingEvent(source string, pay
 	return service.createEvent(events.EventTypeMessagePhoneSending, source, payload)
 }
 
+func (service *MessageService) createMessagePhoneSentEvent(source string, payload events.MessagePhoneSentPayload) (cloudevents.Event, error) {
+	return service.createEvent(events.EventTypeMessagePhoneSent, source, payload)
+}
+
 func (service *MessageService) createEvent(eventType string, source string, payload any) (cloudevents.Event, error) {
 	event := cloudevents.NewEvent()
 

api/pkg/validators/message_handler_validator.go

@@ -115,3 +115,24 @@ func (validator MessageHandlerValidator) ValidateMessageIndex(_ context.Context,
 	})
 	return v.ValidateStruct()
 }
+
+// ValidateMessageEvent validates the requests.MessageEvent request
+func (validator MessageHandlerValidator) ValidateMessageEvent(_ context.Context, request requests.MessageEvent) url.Values {
+	v := govalidator.New(govalidator.Options{
+		Data: &request,
+		Rules: govalidator.MapData{
+			"sent_at": []string{
+				"required",
+			},
+			"event_name": []string{
+				"required",
+				"in:SENT",
+			},
+			"messageID": []string{
+				"required",
+				"uuid",
+			},
+		},
+	})
+	return v.ValidateStruct()
+}