Implementing Real-Time Messaging with Socket.io Send, Receive, and Broadcast

Implementing Real-Time Messaging with Socket.io

This guide covers the core messaging implementation sending, receiving, and broadcasting messages in real-time.

Server-Side Implementation

// src/config/socketEvents.js
const Message = require("../models/Message");
const Conversation = require("../models/Conversation");
const onlineUsers = new Map();

const getRoomId = (userId1, userId2) => {
  return [userId1, userId2].sort().join("_");
};

module.exports = (io) => {
  // JWT authentication
  io.use(async (socket, next) => {
    const token = socket.handshake.auth.token;
    const decoded = jwt.verify(token, process.env.JWT_SECRET);
    socket.user = await User.findById(decoded._id);
    socket.userId = socket.user._id.toString();
    next();
  });

  io.on("connection", (socket) => {
    console.log(`Connected: ${socket.user.firstName}`);
    onlineUsers.set(socket.userId, socket.id);
    io.emit("user:online", { userId: socket.userId });

    // Join a chat room
    socket.on("join:chat", ({ targetUserId }) => {
      const roomId = getRoomId(socket.userId, targetUserId);
      socket.join(roomId);
      socket.emit("room:joined", { roomId, targetUserId });
    });

    // Send a message
    socket.on("message:send", async ({ targetUserId, text }, callback) => {
      try {
        // Validate
        if (!targetUserId || !text) {
          return callback({ success: false, error: "Missing required fields" });
        }
        if (text.length > 1000) {
          return callback({ success: false, error: "Message too long" });
        }

        // Save to database
        const message = await Message.create({
          senderId: socket.userId,
          receiverId: targetUserId,
          text
        });

        // Update conversation
        await Conversation.findOneAndUpdate(
          { participants: { $all: [socket.userId, targetUserId] } },
          {
            $set: { 
              lastMessage: message._id, 
              lastMessageAt: new Date() 
            },
            $inc: { [`unreadCount.${targetUserId}`]: 1 }
          },
          { upsert: true }
        );

        // Broadcast to room
        const roomId = getRoomId(socket.userId, targetUserId);
        const messageData = {
          _id: message._id,
          senderId: socket.userId,
          receiverId: targetUserId,
          text: message.text,
          createdAt: message.createdAt,
          sender: {
            firstName: socket.user.firstName,
            lastName: socket.user.lastName,
            photoUrl: socket.user.photoUrl
          }
        };

        io.to(roomId).emit("message:received", messageData);

        // Check if receiver is online
        const receiverSocketId = onlineUsers.get(targetUserId);
        if (!receiverSocketId) {
          // Receiver is offline   send email notification
          const receiver = await User.findById(targetUserId);
          if (receiver && !receiver.emailBounced) {
            sendEmail(receiver.email, "newMessage", {
              senderFirstName: socket.user.firstName,
              message: text.substring(0, 100),
              chatLink: `https://app.yourdomain.com/chat/${socket.userId}`
            }).catch(err => console.error("Email notification failed:", err));
          }
        }

        // Acknowledge success
        callback({ success: true, message: messageData });
      } catch (err) {
        console.error("Message send error:", err);
        callback({ success: false, error: "Failed to send message" });
      }
    });

    // Mark messages as read
    socket.on("messages:read", async ({ targetUserId }) => {
      await Message.updateMany(
        { senderId: targetUserId, receiverId: socket.userId, read: false },
        { $set: { read: true, readAt: new Date() } }
      );

      // Reset unread count
      await Conversation.findOneAndUpdate(
        { participants: { $all: [socket.userId, targetUserId] } },
        { $set: { [`unreadCount.${socket.userId}`]: 0 } }
      );

      // Notify sender that messages were read
      const roomId = getRoomId(socket.userId, targetUserId);
      io.to(roomId).emit("messages:read", { 
        readBy: socket.userId, 
        targetUserId 
      });
    });

    // Disconnect
    socket.on("disconnect", () => {
      onlineUsers.delete(socket.userId);
      io.emit("user:offline", { userId: socket.userId });
    });
  });
};

Client-Side Implementation

// src/components/Chat/ChatWindow.jsx
import { useEffect, useState, useRef } from "react";
import { socket } from "../../utils/socket";

const ChatWindow = ({ targetUserId, targetUser }) => {
  const [messages, setMessages] = useState([]);
  const [newMessage, setNewMessage] = useState("");
  const [isSending, setIsSending] = useState(false);
  const messagesEndRef = useRef(null);

  // Load message history
  useEffect(() => {
    const loadHistory = async () => {
      const res = await fetch(`/api/chat/${targetUserId}`);
      const data = await res.json();
      setMessages(data.data);
      
      // Join Socket.io room
      socket.emit("join:chat", { targetUserId });
      
      // Mark as read
      socket.emit("messages:read", { targetUserId });
    };
    loadHistory();
  }, [targetUserId]);

  // Listen for new messages
  useEffect(() => {
    const handleMessage = (message) => {
      setMessages((prev) => [...prev, message]);
      // Auto-scroll to bottom
      messagesEndRef.current?.scrollIntoView({ behavior: "smooth" });
      
      // Mark as read if the message is from the target user
      if (message.senderId === targetUserId) {
        socket.emit("messages:read", { targetUserId });
      }
    };

    socket.on("message:received", handleMessage);
    return () => socket.off("message:received", handleMessage);
  }, [targetUserId]);

  // Send message
  const handleSend = async () => {
    if (!newMessage.trim() || isSending) return;
    setIsSending(true);

    socket.emit("message:send", 
      { targetUserId, text: newMessage },
      (response) => {
        setIsSending(false);
        if (response.success) {
          setNewMessage(""); // Clear input
        } else {
          console.error("Failed to send:", response.error);
        }
      }
    );
  };

  return (
    <div className="chat-window">
      <div className="messages">
        {messages.map((msg) => (
          <div key={msg._id} className={msg.senderId === targetUserId ? "received" : "sent"}>
            {msg.text}
          </div>
        ))}
        <div ref={messagesEndRef} />
      </div>
      <div className="input-area">
        <input
          value={newMessage}
          onChange={(e) => setNewMessage(e.target.value)}
          onKeyPress={(e) => e.key === "Enter" && handleSend()}
          placeholder="Type a message..."
        />
        <button onClick={handleSend} disabled={isSending}>
          Send
        </button>
      </div>
    </div>
  );
};

The Takeaway

Real-time messaging with Socket.io involves: server-side event handlers for message:send (save to DB, broadcast to room, send email if offline, acknowledge), join:chat (join room), messages:read (mark as read, notify sender), and disconnect (update online status). Client-side: emit message:send with callback acknowledgment, listen for message:received, auto-scroll, and mark messages as read. Use acknowledgments to confirm successful delivery.