Group Chats

await rabbitmqService.consumeFromSubscriptionQueue(async (message, ack) => {
  try {
    const { userId, groupIds, channelIds } = message

    const socketId = chatsStore.getClient(userId)
    if (socketId) {
      const socket = io.sockets.sockets.get(socketId)
      if (socket) {
        // Bind each channel to this server's queue
        const bindChannelPromises = []
        for (const channelId of channelIds) {
          bindChannelPromises.push(rabbitmqService.bindChannelToQueue(channelId))
        }
        await Promise.all(bindChannelPromises)

        // Join Socket.IO rooms for each channel
        socket.join(channelIds.map(id => id.toString()))

        // Store channels and groups in chats store with reference counting
        chatsStore.addUserChannels(userId, channelIds)
        chatsStore.addUserGroups(userId, groupIds)

        // ...
      }
    }

    ack()
  } catch (error) {
    // ...
  }
})

const tempMessage = {
  hash: generateTempHash(),
  channelId: 2,
  content: 'Hello everyone!',
  status: MessageStatus.SENDING,
}
addGroupMessage(channelId, tempMessage)

await sendGroupMessage(groupId, channelId, 'Hello everyone!', hash)

func (s *GroupService) SendGroupMessage(ctx context.Context, userID int64, dto *SendGroupMessageDTO) error {
  // Verify user has access to this group
  if err := s.verifyGroupAccess(ctx, dto.GroupID, userID); err != nil {
    return err
  }

  // Publish to incoming exchange for worker to process
  message := RabbitMQMessage{
    Type: "MESSAGE_SEND",
    Payload: map[string]any{
      "senderId":   userID,
      "groupId":    dto.GroupID,
      "channelId":  dto.ChannelID,
      "content":    dto.Content,
      "hash":       dto.Hash,
    },
  }

  return s.rabbitmqService.PublishToIncoming("group.message", message)
}

func (mp *MessageProcessor) ProcessGroupMessage(payload *broker.GroupMessagePayload) error {
    return mp.db.Transaction(func(tx *gorm.DB) error {
        // Fetch sender and channel
        var sender domain.UserProfile
        if err := tx.First(&sender, payload.SenderId).Error; err != nil {
            return fmt.Errorf("sender not found: %w", err)
        }

        var channel domain.Channel
        if err := tx.First(&channel, payload.ChannelId).Error; err != nil {
            return fmt.Errorf("channel not found: %w", err)
        }

        // Get all members of the group except the sender
        var userGroups []domain.UserGroup
        if err := tx.Preload("User").Where("group_id = ? AND user_id != ?", payload.GroupId, payload.SenderId).
            Find(&userGroups).Error; err != nil {
            return fmt.Errorf("failed to fetch group members: %w", err)
        }

        // Save the main message record
        message := &domain.Message{
            SenderID:  payload.SenderId,
            Sender:    &sender,
            ChannelID: &payload.ChannelId,
            Channel:   &channel,
            Content:   payload.Content,
        }
        if err := tx.Create(message).Error; err != nil {
            return fmt.Errorf("failed to create message: %w", err)
        }

        // Create message recipients for all group members
        for _, ug := range userGroups {
            recipient := &domain.MessageRecipient{
                MessageID:  message.ID,
                Message:    message,
                ReceiverID: ug.UserID,
                Receiver:   ug.User,
                Status:     domain.MessageStatusSent,
            }
            if err := tx.Create(recipient).Error; err != nil {
                log.Printf("failed to create recipient for user %d: %v", ug.UserID, err)
            }
        }

        // Publish to the channel. RabbitMQ pushes the event to all queues which have binding for this channelId
        routingKey := "channel:" + strconv.FormatInt(payload.ChannelId, 10)
        if err := mp.broker.PublishToOutgoing(routingKey, map[string]any{
            "event":     "group:receive-message",
            "channelId": payload.ChannelId,
            "data": map[string]any{
                "channelId": payload.ChannelId,
                "groupId":   payload.GroupId,
                "messageId": message.ID,
                "content":   payload.Content,
                "senderId":  payload.SenderId,
                "createdAt": message.CreatedAt,
                "status":    domain.MessageStatusSent,
            },
        }); err != nil {
            log.Printf("failed to publish message to channel %d: %v", payload.ChannelId, err)
        }

        return nil
    })
}

await rabbitmqService.consumeFromServerQueue((msg, ack) => {
  try {
    const { event, userId, channelId, data } = msg;

    if (userId) {
      // Direct delivery to users (for personal chats)
      const socketId = chatsStore.getClient(userId);
      if (socketId) io.to(socketId).emit(event, data);
    }
    else if (channelId) {
      // Broadcast to all local sockets in the room (for group channels)
      io.to(channelId.toString()).emit(event, data);
    }

    ack()
  } catch (error) {
    console.error('Error processing message from RabbitMQ server queue:', error)
    ack()
  }
});

socket.on('GROUP_MESSAGE_SENT', (data) => {
  // Replace temp message with actual message
  messages.value = messages.value.map(msg =>
    msg.hash === data.hash ? { ...msg, id: data.messageId, status: MessageStatus.SENT } : msg
  )
})

socket.on('GROUP_MESSAGE_RECEIVE', (data) => {
  messages.value.push({
    id: data.messageId,
    senderId: data.senderId,
    content: data.content,
    createdAt: data.createdAt,
    status: MessageStatus.SENT,
  })
})