send events to the room

internal/chat/application/dto/send_room_event_dto.go

@@ -0,0 +1,8 @@
+package dto
+
+// SendRoomEventReqDTO is the request DTO for sending room events
+type SendRoomEventReqDTO struct {
+	RoomID    string `validate:"required"`
+	UserID    string `validate:"required"`
+	EventType string `json:"event_type" validate:"required"`
+}

internal/chat/application/room_command_handler.go

@@ -12,6 +12,7 @@ type RoomCommandHandler interface {
 	CreateRoom(*dto.CreateRoomReqDTO) (*values.RoomValue, *errorhandler.Response)
 	JoinRoom(*dto.JoinRoomReqDTO) (*values.RoomValue, *errorhandler.Response)
 	LeaveRoom(*dto.LeaveRoomReqDTO) (*values.RoomValue, *errorhandler.Response)
+	SendRoomEvent(*dto.SendRoomEventReqDTO) *errorhandler.Response
 }
 
 type roomCommandHandler struct {
@@ -37,3 +38,7 @@ func (r *roomCommandHandler) JoinRoom(req *dto.JoinRoomReqDTO) (*values.RoomValu
 func (r *roomCommandHandler) LeaveRoom(req *dto.LeaveRoomReqDTO) (*values.RoomValue, *errorhandler.Response) {
 	return r.roomCommandServices.LeaveRoom(req)
 }
+
+func (r *roomCommandHandler) SendRoomEvent(req *dto.SendRoomEventReqDTO) *errorhandler.Response {
+	return r.roomCommandServices.SendRoomEvent(req)
+}

internal/chat/domain/entity/room.go

@@ -4,6 +4,13 @@ import (
 	"time"
 )
 
+type RoomEventType string
+
+const (
+	RoomEventTypeTypingStarted RoomEventType = "typing_started"
+	RoomEventTypeTypingEnded   RoomEventType = "typing_ended"
+)
+
 type Room struct {
 	id string
 

internal/chat/domain/services/room_command_services.go

@@ -21,6 +21,7 @@ type RoomCommandDomainServices interface {
 	UpdateLastMessage(string, *values.MessageValue) *errorhandler.Response
 	JoinRoom(*dto.JoinRoomReqDTO) (*values.RoomValue, *errorhandler.Response)
 	LeaveRoom(*dto.LeaveRoomReqDTO) (*values.RoomValue, *errorhandler.Response)
+	SendRoomEvent(*dto.SendRoomEventReqDTO) *errorhandler.Response
 }
 
 type roomCommandDomainServices struct {
@@ -176,3 +177,48 @@ func (r *roomCommandDomainServices) LeaveRoom(req *dto.LeaveRoomReqDTO) (*values
 
 	return values.NewRoomValueFromRoom(roomEntity), nil
 }
+
+// SendRoomEvent sends an event to a room
+func (r *roomCommandDomainServices) SendRoomEvent(req *dto.SendRoomEventReqDTO) *errorhandler.Response {
+	// Query the room
+	room, err := r.roomRepository.GetRoomByID(req.RoomID)
+
+	if err != nil {
+		return err
+	}
+
+	if room == nil {
+		return &errorhandler.Response{Code: errorhandler.RoomNotFoundErrorCode, Message: errorhandler.RoomNotFoundMessage, StatusCode: fiber.StatusNotFound}
+	}
+
+	// Convert the room to entity
+	roomEntity := room.ToRoom()
+
+	// Check if the user is in the room
+	if !roomEntity.CheckRoomUserExists(req.UserID) {
+		return &errorhandler.Response{Code: errorhandler.UserIsNotInRoomCode, Message: errorhandler.UserIsNotInRoomMessage, StatusCode: fiber.StatusBadRequest}
+	}
+
+	// Publishes to user websocket
+	websocketEventValue := websocketValues.RoomNewEventWebsocketValue{
+		SenderID:  req.UserID,
+		RoomID:    roomEntity.GetID(),
+		UserIDs:   []string{},
+		EventType: req.EventType,
+	}
+
+	// Loop users and add them to the event with userid
+	for _, user := range roomEntity.GetRoomUsers() {
+		websocketEventValue.UserIDs = append(websocketEventValue.UserIDs, user.UserID)
+	}
+
+	messageJSON, _ := json.Marshal(websocketEventValue)
+
+	_, publishErr := r.ctx.GetNatsContext().GetJetStreamContext().Publish(websocketTypes.RoomEvents, messageJSON)
+
+	if publishErr != nil {
+		r.ctx.GetLogger().Error(publishErr.Error())
+	}
+
+	return nil
+}

internal/chat/interfaces/http/room.go

@@ -60,7 +60,7 @@ func (h *handler) createRoom() fiber.Handler {
 
 		if resErr != nil {
 			h.application.AppContext.GetLogger().Logger.Warn(
-				errorhandler.InvalidCredentialsMessage,
+				resErr.Message.(string),
 				zap.Error(err),
 			)
 
@@ -131,7 +131,7 @@ func (h *handler) queryRooms() fiber.Handler {
 
 		if resErr != nil {
 			h.application.AppContext.GetLogger().Logger.Warn(
-				errorhandler.InvalidCredentialsMessage,
+				resErr.Message.(string),
 				zap.Error(err),
 			)
 
@@ -183,7 +183,7 @@ func (h *handler) joinRoom() fiber.Handler {
 
 		if resErr != nil {
 			h.application.AppContext.GetLogger().Logger.Warn(
-				errorhandler.InvalidCredentialsMessage,
+				resErr.Message.(string),
 				zap.Error(err),
 			)
 
@@ -235,7 +235,69 @@ func (h *handler) leaveRoom() fiber.Handler {
 
 		if resErr != nil {
 			h.application.AppContext.GetLogger().Logger.Warn(
-				errorhandler.InvalidCredentialsMessage,
+				resErr.Message.(string),
+				zap.Error(err),
+			)
+
+			return f.Status(resErr.StatusCode).JSON(resErr)
+		}
+
+		return f.Status(fiber.StatusOK).JSON(nil)
+	}
+}
+
+// sendRoomEvent creates a new event in a chat room.
+// @Summary Sends a room event
+// @Description Allows authenticated users to send a new event in a chat room.
+// @Tags Room
+// @Accept  json
+// @Produce  json
+// @Security ApiKeyAuth
+// @Param   roomID path string true "Room ID"
+// @Param   body body string true "Event body"
+// @Success 200 {object} dto.SendRoomEventReqDTO
+// @Failure 400 {object} errorhandler.Response
+// @Router /chat/room/{roomID}/event [post]
+func (h *handler) sendRoomEvent() fiber.Handler {
+	return func(f *fiber.Ctx) error {
+		// Serialize the body
+		var request dto.SendRoomEventReqDTO
+		err := f.BodyParser(&request)
+		if err != nil {
+			h.application.AppContext.GetLogger().Logger.Warn(
+				errorhandler.RequestBodyParseErrorMessage,
+				zap.Error(err),
+			)
+
+			return f.Status(fiber.StatusBadRequest).JSON(&errorhandler.Response{Code: errorhandler.RequestBodyParseErrorCode, Message: errorhandler.RequestBodyParseErrorMessage, StatusCode: fiber.StatusBadRequest})
+		}
+
+		request.RoomID = f.Params("roomID")
+		request.UserID = f.Locals("userID").(string)
+
+		validate := validator.New()
+
+		// Validate the request
+		err = validate.Struct(request)
+		if err != nil {
+			h.application.AppContext.GetLogger().Logger.Warn(
+				errorhandler.ValidationErrorMessage,
+				zap.Error(err),
+			)
+
+			fields := []string{}
+			for _, err := range err.(validator.ValidationErrors) {
+				fields = append(fields, err.Field())
+			}
+			return f.Status(fiber.StatusBadRequest).JSON(&errorhandler.Response{Code: errorhandler.ValidationErrorCode, Message: fmt.Sprintf("invalid fields %s", fields), StatusCode: fiber.StatusBadRequest})
+		}
+
+		// Handle the request
+		resErr := h.application.RoomCommandHandler.SendRoomEvent(&request)
+
+		if resErr != nil {
+			h.application.AppContext.GetLogger().Logger.Warn(
+				resErr.Message.(string),
 				zap.Error(err),
 			)
 

internal/chat/interfaces/http/router.go

@@ -29,6 +29,7 @@ func (h *handler) RegisterRoutes(f fiber.Router) {
 		chat.Get("/rooms", h.queryRooms())
 		chat.Post("/room/:roomID/join", h.joinRoom())
 		chat.Post("/room/:roomID/leave", h.leaveRoom())
+		chat.Post("/room/:roomID/event", h.sendRoomEvent())
 
 		// Message routes
 		chat.Post("/room/:roomID/message", h.createMessage())

internal/user/domain/event/types/types.go

@@ -8,3 +8,4 @@ const StreamSubjects = "USER.*"
 const MessageCreatedEvent = "USER.MessageCreated"
 const RoomUserJoinedEvent = "USER.RoomUserJoined"
 const RoomUserLeftEvent = "USER.RoomUserLeft"
+const RoomEvents = "USER.RoomEvents"

internal/user/domain/values/websocket/room_new_event.go

@@ -0,0 +1,8 @@
+package websocket
+
+type RoomNewEventWebsocketValue struct {
+	RoomID    string   `json:"room_id"`
+	SenderID  string   `json:"sender_id"`
+	UserIDs   []string `json:"user_ids"`
+	EventType string   `json:"event"`
+}

internal/user/interfaces/ws/listeners.go

@@ -22,6 +22,8 @@ func (h *handler) setupListeners() {
 			return h.onMessageCreated(msg)
 		case eventTypes.RoomUserJoinedEvent:
 			return h.onUserJoinedRoom(msg)
+		case eventTypes.RoomEvents:
+			return h.onRoomEvent(msg)
 		}
 
 		return nil
@@ -114,3 +116,50 @@ func (h *handler) onUserJoinedRoom(msg *nats.Msg) error {
 
 	return nil
 }
+
+// onRoomEvent handles the room event
+func (h *handler) onRoomEvent(msg *nats.Msg) error {
+	msg.Ack()
+
+	// Unmarshal the message
+	var event websocketValues.RoomNewEventWebsocketValue
+	err := json.Unmarshal(msg.Data, &event)
+
+	if err != nil {
+		return err
+	}
+
+	// Check if the user ID is in clients
+	h.clients.Range(func(client, v interface{}) bool {
+		nc := v.(*websocket.Conn)
+
+		// Range the users
+		for _, user := range event.UserIDs {
+			// If the senderID is the same as the client ID, skip
+			if nc.UserValue("ID").(string) == event.SenderID {
+				return true
+			}
+
+			// If the user ID is in the connected clients, send the event
+			if nc.UserValue("ID").(string) == user {
+				h.application.AppContext.GetLogger().Logger.Info("Sending room event to ws client with ID: ", zap.String("ID", nc.UserValue("ID").(string)))
+
+				// Create the event model
+				eventModel := &types.WebsocketMessage{}
+				eventModel.New(types.MessageTypeRoomEvent)
+				eventModel.ConnectionID = fmt.Sprintf("%d", nc.ID())
+				eventModel.Message = map[string]string{"room_id": event.RoomID, "user_id": event.SenderID, "event_type": event.EventType}
+
+				nc.Write(eventModel.ToJson())
+
+				return true
+
+			}
+
+		}
+
+		return true
+	})
+
+	return nil
+}

internal/user/interfaces/ws/types/events.go

@@ -15,4 +15,5 @@ const (
 	// Room
 	MessageTypeRoomUserJoined = "room.user.joined"
 	MessageTypeRoomUserLeft   = "room.user.left"
+	MessageTypeRoomEvent      = "room.event"
 )