Fix outstanding messages

Author: AchoArnold

api/pkg/entities/heartbeat.go

@@ -12,5 +12,5 @@ type Heartbeat struct {
 	Owner     string    `json:"owner" gorm:"index:idx_heartbeats_owner_timestamp" example:"+18005550199"`
 	UserID    UserID    `json:"userID" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
 	Timestamp time.Time `json:"timestamp" gorm:"index:idx_heartbeats_owner_timestamp" example:"2022-06-05T14:26:01.520828+03:00"`
-	Quantity  int       `json:"quantity" example:"2"`
+	MessageID uuid.UUID `json:"message_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
 }

api/pkg/events/message_phone_sending_event.go

@@ -1,6 +1,8 @@
 package events
 
 import (
+	"time"
+
 	"github.com/NdoleStudio/http-sms-manager/pkg/entities"
 	"github.com/google/uuid"
 )
@@ -10,9 +12,10 @@ const EventTypeMessagePhoneSending = "message.phone.sending"
 
 // MessagePhoneSendingPayload is the payload of the EventTypeMessageSent event
 type MessagePhoneSendingPayload struct {
-	ID      uuid.UUID       `json:"id"`
-	UserID  entities.UserID `json:"user_id"`
-	Owner   string          `json:"owner"`
-	Contact string          `json:"contact"`
-	Content string          `json:"content"`
+	ID        uuid.UUID       `json:"id"`
+	UserID    entities.UserID `json:"user_id"`
+	Timestamp time.Time       `json:"timestamp"`
+	Owner     string          `json:"owner"`
+	Contact   string          `json:"contact"`
+	Content   string          `json:"content"`
 }

api/pkg/handlers/message_handler.go

@@ -93,20 +93,19 @@ func (h *MessageHandler) PostSend(c *fiber.Ctx) error {
 	return h.responseOK(c, "message added to queue", message)
 }
 
-// GetOutstanding returns entities.Message which are still to be sent by the mobile phone
-// @Summary      Get outstanding messages
-// @Description  Get list of messages which are outstanding to be sent by the phone
+// GetOutstanding returns an entities.Message which is still to be sent by the mobile phone
+// @Summary      Get an outstanding message
+// @Description  Get an outstanding message to be sent by an android phone
 // @Security	 ApiKeyAuth
 // @Tags         Messages
 // @Accept       json
 // @Produce      json
-// @Param        owner	query  string  	true 							"The owner's phone number" 					default(+18005550199)
-// @Param        limit	query  int  	false  							"Number of outstanding messages to fetch"	minimum(1)	maximum(10)
-// @Success      200 	{object}		responses.MessagesResponse
-// @Failure      400	{object}		responses.BadRequest
+// @Param        message_id	query  		string  						true "The ID of the message" default(32343a19-da5e-4b1b-a767-3298a73703cb)
+// @Success      200 		{object}	responses.MessageResponse
+// @Failure      400		{object}	responses.BadRequest
 // @Failure 	 401    	{object}	responses.Unauthorized
-// @Failure      422	{object}		responses.UnprocessableEntity
-// @Failure      500	{object}		responses.InternalServerError
+// @Failure      422		{object}	responses.UnprocessableEntity
+// @Failure      500		{object}	responses.InternalServerError
 // @Router       /messages/outstanding [get]
 func (h *MessageHandler) GetOutstanding(c *fiber.Ctx) error {
 	ctx, span := h.tracer.StartFromFiberCtx(c)
@@ -128,14 +127,14 @@ func (h *MessageHandler) GetOutstanding(c *fiber.Ctx) error {
 		return h.responseUnprocessableEntity(c, errors, "validation errors while fetching outstanding messages")
 	}
 
-	messages, err := h.service.GetOutstanding(ctx, request.ToGetOutstandingParams(c.OriginalURL(), h.userIDFomContext(c), timestamp))
+	message, err := h.service.GetOutstanding(ctx, request.ToGetOutstandingParams(c.OriginalURL(), h.userIDFomContext(c), timestamp))
 	if err != nil {
-		msg := fmt.Sprintf("cannot get messgaes with URL [%s]", c.OriginalURL())
+		msg := fmt.Sprintf("cannot get outstnading messgage with ID [%s]", request.MessageID)
 		ctxLogger.Error(stacktrace.Propagate(err, msg))
 		return h.responseInternalServerError(c)
 	}
 
-	return h.responseOK(c, fmt.Sprintf("fetched %d %s", len(*messages), h.pluralize("message", len(*messages))), messages)
+	return h.responseOK(c, "outstanding message fetched successfully", message)
 }
 
 // Index returns messages sent between 2 phone numbers

api/pkg/listeners/heartbeat_listener.go

@@ -31,16 +31,16 @@ func NewHeartbeatListener(
 	}
 
 	return l, map[string]events.EventListener{
-		events.EventTypeHeartbeatPhoneOutstanding: l.onHeartbeatPhoneOutstanding,
+		events.EventTypeMessagePhoneSending: l.onMessagePhoneSending,
 	}
 }
 
-// onHeartbeatPhoneOutstanding handles the events.EventTypeHeartbeatPhoneOutstanding event
-func (listener *HeartbeatListener) onHeartbeatPhoneOutstanding(ctx context.Context, event cloudevents.Event) error {
+// onMessagePhoneSending handles the events.EventTypeMessagePhoneSending event
+func (listener *HeartbeatListener) onMessagePhoneSending(ctx context.Context, event cloudevents.Event) error {
 	ctx, span := listener.tracer.Start(ctx)
 	defer span.End()
 
-	var payload events.HeartbeatPhoneOutstandingPayload
+	var payload events.MessagePhoneSendingPayload
 	if err := event.DataAs(&payload); err != nil {
 		msg := fmt.Sprintf("cannot decode [%s] into [%T]", event.Data(), payload)
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
@@ -49,7 +49,8 @@ func (listener *HeartbeatListener) onHeartbeatPhoneOutstanding(ctx context.Conte
 	storeParams := services.HeartbeatStoreParams{
 		Owner:     payload.Owner,
 		Timestamp: payload.Timestamp,
-		Quantity:  payload.Quantity,
+		UserID:    payload.UserID,
+		MessageID: payload.ID,
 	}
 
 	if _, err := listener.service.Store(ctx, storeParams); err != nil {

api/pkg/repositories/gorm_heartbeat_repository.go

@@ -31,11 +31,11 @@ func NewGormHeartbeatRepository(
 }
 
 // Index entities.Message between 2 parties
-func (repository *gormHeartbeatRepository) Index(ctx context.Context, owner string, params IndexParams) (*[]entities.Heartbeat, error) {
+func (repository *gormHeartbeatRepository) Index(ctx context.Context, userID entities.UserID, owner string, params IndexParams) (*[]entities.Heartbeat, error) {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
 
-	query := repository.db.WithContext(ctx).Where("owner = ?", owner)
+	query := repository.db.WithContext(ctx).Where("user_id = ?", userID).Where("owner = ?", owner)
 	if len(params.Query) > 0 {
 		queryPattern := "%" + params.Query + "%"
 		query.Where("quantity ILIKE ?", queryPattern)

api/pkg/repositories/gorm_message_repository.go

@@ -106,36 +106,25 @@ func (repository *gormMessageRepository) Update(ctx context.Context, message *en
 }
 
 // GetOutstanding fetches messages that still to be sent to the phone
-func (repository *gormMessageRepository) GetOutstanding(ctx context.Context, userID entities.UserID, owner string, take int) (*[]entities.Message, error) {
+func (repository *gormMessageRepository) GetOutstanding(ctx context.Context, userID entities.UserID, messageID uuid.UUID) (*entities.Message, error) {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
 
-	messages := new([]entities.Message)
+	message := new(entities.Message)
 	err := crdbgorm.ExecuteTx(ctx, repository.db, nil,
 		func(tx *gorm.DB) error {
-			return tx.WithContext(ctx).Model(messages).
+			return tx.WithContext(ctx).Model(message).
 				Clauses(clause.Returning{}).
 				Where("user_id = ?", userID).
-				Where(
-					"id IN (?)",
-					tx.Model(&entities.Message{}).
-						Where("owner = ?", owner).
-						Where(
-							repository.db.
-								Where("status = ?", entities.MessageStatusPending).
-								Where("age(now(), created_at) < ?", "5 minutes"),
-						).
-						Order("request_received_at ASC").
-						Select("id").
-						Limit(take),
-				).
-				Update("status", "sending").Error
+				Where("id = ?", messageID).
+				Where("status = ?", entities.MessageStatusPending).
+				Update("status", entities.MessageStatusSending).Error
 		},
 	)
 	if err != nil {
-		msg := fmt.Sprintf("cannot fetch [%d] outstanding messages", take)
+		msg := fmt.Sprintf("cannot fetch outstanding message with userID [%s] and messageID [%s]", userID, messageID)
 		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
 	}
 
-	return messages, nil
+	return message, nil
 }

api/pkg/repositories/heartbeat_repository.go

@@ -12,5 +12,5 @@ type HeartbeatRepository interface {
 	Store(ctx context.Context, heartbeat *entities.Heartbeat) error
 
 	// Index entities.Heartbeat of an owner
-	Index(ctx context.Context, owner string, params IndexParams) (*[]entities.Heartbeat, error)
+	Index(ctx context.Context, userID entities.UserID, owner string, params IndexParams) (*[]entities.Heartbeat, error)
 }

api/pkg/repositories/message_repository.go

@@ -21,6 +21,6 @@ type MessageRepository interface {
 	// Index entities.Message between 2 phone numbers
 	Index(ctx context.Context, userID entities.UserID, owner string, contact string, params IndexParams) (*[]entities.Message, error)
 
-	// GetOutstanding fetches list of outstanding []entities.Message
-	GetOutstanding(ctx context.Context, userID entities.UserID, owner string, limit int) (*[]entities.Message, error)
+	// GetOutstanding fetches an entities.Message which is outstanding
+	GetOutstanding(ctx context.Context, userID entities.UserID, messageID uuid.UUID) (*entities.Message, error)
 }

api/pkg/requests/message_outstanding_request.go

@@ -1,33 +1,23 @@
 package requests
 
 import (
-	"strconv"
 	"strings"
 	"time"
 
 	"github.com/NdoleStudio/http-sms-manager/pkg/entities"
-
 	"github.com/NdoleStudio/http-sms-manager/pkg/services"
+	"github.com/google/uuid"
 )
 
 // MessageOutstanding is the payload fetching outstanding entities.Message
 type MessageOutstanding struct {
 	request
-	Owner string `json:"owner" query:"owner"`
-	Limit string `json:"limit" query:"limit"`
+	MessageID string `json:"message_id" query:"message_id"`
 }
 
 // Sanitize sets defaults to MessageOutstanding
 func (input *MessageOutstanding) Sanitize() MessageOutstanding {
-	if strings.TrimSpace(input.Limit) == "" {
-		input.Limit = "1"
-	}
-
-	input.Owner = input.sanitizeAddress(input.Owner)
-	if input.Owner == "" {
-		input.Owner = "+37259139660"
-	}
-
+	input.MessageID = strings.TrimSpace(input.MessageID)
 	return *input
 }
 
@@ -36,14 +26,7 @@ func (input *MessageOutstanding) ToGetOutstandingParams(source string, userID en
 	return services.MessageGetOutstandingParams{
 		Source:    source,
 		UserID:    userID,
-		Owner:     input.Owner,
+		MessageID: uuid.MustParse(input.MessageID),
 		Timestamp: timestamp,
-		Limit:     input.getLimit(),
 	}
 }
-
-// getLimit gets the take as a string
-func (input *MessageOutstanding) getLimit() int {
-	val, _ := strconv.Atoi(input.Limit)
-	return val
-}

api/pkg/services/heartbeat_service.go

@@ -35,13 +35,13 @@ func NewHeartbeatService(
 }
 
 // Index fetches the heartbeats for a phone number
-func (service *HeartbeatService) Index(ctx context.Context, owner string, params repositories.IndexParams) (*[]entities.Heartbeat, error) {
+func (service *HeartbeatService) Index(ctx context.Context, userID entities.UserID, owner string, params repositories.IndexParams) (*[]entities.Heartbeat, error) {
 	ctx, span := service.tracer.Start(ctx)
 	defer span.End()
 
 	ctxLogger := service.tracer.CtxLogger(service.logger, span)
 
-	heartbeats, err := service.repository.Index(ctx, owner, params)
+	heartbeats, err := service.repository.Index(ctx, userID, owner, params)
 	if err != nil {
 		msg := fmt.Sprintf("could not fetch heartbeats with parms [%+#v]", params)
 		return nil, service.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
@@ -55,7 +55,8 @@ func (service *HeartbeatService) Index(ctx context.Context, owner string, params
 type HeartbeatStoreParams struct {
 	Owner     string
 	Timestamp time.Time
-	Quantity  int
+	UserID    entities.UserID
+	MessageID uuid.UUID
 }
 
 // Store a new entities.Heartbeat
@@ -69,7 +70,8 @@ func (service *HeartbeatService) Store(ctx context.Context, params HeartbeatStor
 		ID:        uuid.New(),
 		Owner:     params.Owner,
 		Timestamp: params.Timestamp,
-		Quantity:  params.Quantity,
+		MessageID: params.MessageID,
+		UserID:    params.UserID,
 	}
 
 	if err := service.repository.Store(ctx, heartbeat); err != nil {