fix: Publish API validation (#272)

* fix: Publish API validate required tenant_id

* fix: Improve topics validation

Author: alexluong

internal/publishmq/eventhandler.go

@@ -18,7 +18,8 @@ import (
 )
 
 var (
-	ErrInvalidTopic = errors.New("invalid topic")
+	ErrInvalidTopic  = errors.New("invalid topic")
+	ErrRequiredTopic = errors.New("topic is required")
 )
 
 type EventHandler interface {
@@ -62,6 +63,9 @@ func NewEventHandler(
 var _ EventHandler = (*eventHandler)(nil)
 
 func (h *eventHandler) Handle(ctx context.Context, event *models.Event) error {
+	if len(h.topics) > 0 && event.Topic == "" {
+		return ErrRequiredTopic
+	}
 	if len(h.topics) > 0 && event.Topic != "*" && !slices.Contains(h.topics, event.Topic) {
 		return ErrInvalidTopic
 	}

internal/services/api/publish_handlers.go

@@ -38,8 +38,24 @@ func (h *PublishHandlers) Ingest(c *gin.Context) {
 	if err := h.eventHandler.Handle(c.Request.Context(), &event); err != nil {
 		if errors.Is(err, idempotence.ErrConflict) {
 			c.Status(http.StatusConflict)
+		} else if errors.Is(err, publishmq.ErrRequiredTopic) {
+			AbortWithValidationError(c, ErrorResponse{
+				Code:    http.StatusUnprocessableEntity,
+				Message: "validation error",
+				Err:     err,
+				Data: map[string]string{
+					"topic": "required",
+				},
+			})
 		} else if errors.Is(err, publishmq.ErrInvalidTopic) {
-			AbortWithValidationError(c, err)
+			AbortWithValidationError(c, ErrorResponse{
+				Code:    http.StatusUnprocessableEntity,
+				Message: "validation error",
+				Err:     err,
+				Data: map[string]string{
+					"topic": "invalid",
+				},
+			})
 		} else {
 			AbortWithError(c, http.StatusInternalServerError, NewErrInternalServer(err))
 		}
@@ -48,10 +64,9 @@ func (h *PublishHandlers) Ingest(c *gin.Context) {
 	c.Status(http.StatusOK)
 }
 
-// TODO: validation
 type PublishedEvent struct {
 	ID               string                 `json:"id"`
-	TenantID         string                 `json:"tenant_id"`
+	TenantID         string                 `json:"tenant_id" binding:"required"`
 	DestinationID    string                 `json:"destination_id"`
 	Topic            string                 `json:"topic"`
 	EligibleForRetry bool                   `json:"eligible_for_retry"`