📋 What We're Building
- ✅ Real-time messaging between users
- ✅ JWT authentication for secure connections
- ✅ MongoDB for message persistence
- ✅ Typing indicators and read receipts
- ✅ Conversation history with pagination
- ✅ Unread message count
🏗️ Architecture
┌─────────────────────────────────────────────────────────────────────────────┐
│ P2P MESSAGING SYSTEM │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ CLIENT (Mobile / Web) │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Send │ │ Receive │ │ Read │ │ │
│ │ │ Message │ │ Message │ │ Receipts │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ │ │ │ │ │ │
│ │ └───────────────┼───────────────┘ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────────┐ │ │
│ │ │ SOCKET.IO (WebSocket) │ │ │
│ │ │ • Real-time bidirectional communication │ │ │
│ │ │ • Events: send-message, typing, mark-read │ │ │
│ │ │ • JWT authentication middleware │ │ │
│ │ └─────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────┬───────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ BACKEND API (Node.js/Express) │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────────────┐ ┌─────────────────────┐ │ │
│ │ │ Routes │ │ Controllers │ │ Models │ │ │
│ │ │ /messages │──│ messageController │──│ Message (MongoDB) │ │ │
│ │ └─────────────┘ └─────────────────────┘ └─────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
text
💻 Implementation
1. Install Dependencies
bash
npm install socket.io socket.io-client jsonwebtoken mongoose express
2. WebSocket Server
Create websocket/server.js:
javascript
const socketIO = require('socket.io');
const jwt = require('jsonwebtoken');
const Message = require('../models/Message');
class WebSocketServer {
constructor(server) {
this.io = socketIO(server, {
cors: { origin: '*', methods: ['GET', 'POST'] }
});
this.authenticatedSockets = new Map();
this.setupMiddleware();
this.setupHandlers();
}
setupMiddleware() {
this.io.use(async (socket, next) => {
try {
const token = socket.handshake.auth.token;
if (!token) return next(new Error('Authentication required'));
const decoded = jwt.verify(token, process.env.JWT_SECRET);
const user = await User.findById(decoded.id);
if (!user) return next(new Error('User not found'));
socket.user = user;
next();
} catch (error) {
next(new Error('Invalid token'));
}
});
}
setupHandlers() {
this.io.on('connection', (socket) => {
// Store socket connection
this.authenticatedSockets.set(socket.user._id.toString(), socket.id);
socket.join(`user:${socket.user._id}`);
// Send message
socket.on('send-message', async (data) => {
try {
const { recipientId, content } = data;
const message = new Message({
sender: socket.user._id,
recipient: recipientId,
content,
status: 'sent'
});
await message.save();
await message.populate('sender', 'username fullName');
// Deliver to recipient
const recipientSocketId = this.authenticatedSockets.get(recipientId);
if (recipientSocketId) {
this.io.to(recipientSocketId).emit('new-message', message);
message.status = 'delivered';
await message.save();
}
socket.emit('message-sent', message);
} catch (error) {
socket.emit('message-error', { error: error.message });
}
});
// Typing indicator
socket.on('typing', (data) => {
const { recipientId, isTyping } = data;
const recipientSocketId = this.authenticatedSockets.get(recipientId);
if (recipientSocketId) {
this.io.to(recipientSocketId).emit('user-typing', {
userId: socket.user._id,
username: socket.user.username,
isTyping
});
}
});
// Mark as read
socket.on('mark-read', async (data) => {
const { messageIds } = data;
await Message.updateMany(
{ _id: { $in: messageIds } },
{ status: 'read', readAt: new Date() }
);
});
// Disconnect
socket.on('disconnect', () => {
this.authenticatedSockets.delete(socket.user._id.toString());
});
});
}
}
module.exports = WebSocketServer;
3. Message Model
Create models/Message.js:
javascript
const mongoose = require('mongoose');
const messageSchema = new mongoose.Schema({
sender: {
type: mongoose.Schema.Types.ObjectId,
ref: 'User',
required: true
},
recipient: {
type: mongoose.Schema.Types.ObjectId,
ref: 'User',
required: true
},
order: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Order'
},
content: {
type: String,
required: true,
maxlength: 10000
},
type: {
type: String,
enum: ['text', 'image', 'file', 'system'],
default: 'text'
},
status: {
type: String,
enum: ['sent', 'delivered', 'read', 'failed'],
default: 'sent'
},
readAt: { type: Date },
deliveredAt: { type: Date },
isDeleted: { type: Boolean, default: false }
}, { timestamps: true });
module.exports = mongoose.model('Message', messageSchema);
4. Message Controller
Create controllers/messageController.js:
javascript
const Message = require('../models/Message');
const User = require('../models/User');
exports.sendMessage = async (req, res) => {
try {
const { recipientId, content } = req.body;
const recipient = await User.findById(recipientId);
if (!recipient) {
return res.status(404).json({ success: false, message: 'Recipient not found' });
}
const message = new Message({
sender: req.user._id,
recipient: recipientId,
content,
status: 'sent'
});
await message.save();
await message.populate('sender', 'username fullName');
// Notify via WebSocket
const ws = req.app.get('websocket');
if (ws) {
await ws.notifyUser(recipientId, 'new-message', message);
}
res.status(201).json({ success: true, data: message });
} catch (error) {
res.status(500).json({ success: false, message: error.message });
}
};
exports.getConversation = async (req, res) => {
try {
const { userId } = req.params;
const { limit = 50, before } = req.query;
const otherUser = await User.findById(userId);
if (!otherUser) {
return res.status(404).json({ success: false, message: 'User not found' });
}
const query = {
$or: [
{ sender: req.user._id, recipient: userId },
{ sender: userId, recipient: req.user._id }
],
isDeleted: false
};
if (before) {
query.createdAt = { $lt: new Date(before) };
}
const messages = await Message.find(query)
.sort({ createdAt: -1 })
.limit(parseInt(limit))
.populate('sender', 'username fullName')
.populate('recipient', 'username fullName');
res.json({
success: true,
data: {
messages: messages.reverse(),
user: otherUser,
hasMore: messages.length === parseInt(limit)
}
});
} catch (error) {
res.status(500).json({ success: false, message: error.message });
}
};
exports.getUnreadCount = async (req, res) => {
try {
const count = await Message.countDocuments({
recipient: req.user._id,
status: 'sent'
});
res.json({ success: true, data: { count } });
} catch (error) {
res.status(500).json({ success: false, message: error.message });
}
};
exports.markAsRead = async (req, res) => {
try {
const { messageIds } = req.body;
await Message.updateMany(
{ _id: { $in: messageIds }, recipient: req.user._id },
{ status: 'read', readAt: new Date() }
);
res.json({ success: true, message: 'Messages marked as read' });
} catch (error) {
res.status(500).json({ success: false, message: error.message });
}
};
exports.getConversationsList = async (req, res) => {
try {
const conversations = await Message.aggregate([
{
$match: {
$or: [{ sender: req.user._id }, { recipient: req.user._id }],
isDeleted: false
}
},
{ $sort: { createdAt: -1 } },
{
$group: {
_id: {
$cond: [{ $eq: ['$sender', req.user._id] }, '$recipient', '$sender']
},
lastMessage: { $first: '$$ROOT' },
unreadCount: {
$sum: {
$cond: [
{ $and: [
{ $eq: ['$recipient', req.user._id] },
{ $in: ['$status', ['sent', 'delivered']] }
]},
1,
0
]
}
}
}
},
{
$lookup: {
from: 'users',
localField: '_id',
foreignField: '_id',
as: 'user'
}
},
{ $unwind: '$user' },
{
$project: {
userId: '$_id',
username: '$user.username',
fullName: '$user.fullName',
lastMessage: '$lastMessage.content',
lastMessageAt: '$lastMessage.createdAt',
unreadCount: 1
}
}
]);
res.json({ success: true, data: conversations });
} catch (error) {
res.status(500).json({ success: false, message: error.message });
}
};
5. Routes
Create routes/messages.js:
javascript
const express = require('express');
const router = express.Router();
const messageController = require('../controllers/messageController');
const { authenticate } = require('../middleware/auth');
// All routes require authentication
router.use(authenticate);
// Get conversations list
router.get('/conversations', messageController.getConversationsList);
// Get conversation with user
router.get('/conversation/:userId', messageController.getConversation);
// Send message
router.post('/send', messageController.sendMessage);
// Mark messages as read
router.put('/read', messageController.markAsRead);
// Get unread count
router.get('/unread', messageController.getUnreadCount);
module.exports = router;
6. Integrate in Server
Update server.js:
javascript
const express = require('express');
const http = require('http');
const WebSocketServer = require('./websocket/server');
const app = express();
const server = http.createServer(app);
// Initialize WebSocket server
const ws = new WebSocketServer(server);
app.set('websocket', ws);
// Routes
const messageRoutes = require('./routes/messages');
app.use('/api/messages', messageRoutes);
server.listen(3000, () => {
console.log('🚀 Server running on port 3000');
console.log('✅ WebSocket server initialized');
});
🚀 Testing the System
Test WebSocket Connection
bash
# Install wscat
npm install -g wscat
# Connect to WebSocket
wscat -c ws://localhost:3000
Test API Endpoints
bash
# Get conversations
curl -X GET http://localhost:3000/api/messages/conversations \
-H "Authorization: Bearer YOUR_JWT_TOKEN"
# Send message
curl -X POST http://localhost:3000/api/messages/send \
-H "Authorization: Bearer YOUR_JWT_TOKEN" \
-H "Content-Type: application/json" \
-d '{"recipientId":"USER_ID","content":"Hello!"}'
# Get unread count
curl -X GET http://localhost:3000/api/messages/unread \
-H "Authorization: Bearer YOUR_JWT_TOKEN"
📊 API Reference
Method Endpoint Description
GET /api/messages/conversations Get all conversations
GET /api/messages/conversation/:userId Get conversation with user
POST /api/messages/send Send a message
PUT /api/messages/read Mark messages as read
GET /api/messages/unread Get unread count
🔒 Security Considerations
JWT Authentication — All WebSocket connections and API endpoints are protected
Message Validation — Content length limited to 10,000 characters
Access Control — Users can only access their own conversations
Rate Limiting — Consider adding rate limiting for message sending
📚 Resources
Socket.io Documentation
JWT Authentication Guide
MongoDB Schema Design
🔗 Connect With Me
Platform Link
Telegram Bot @myzubster_bot
Telegram Channel t.me/myzubster
GitHub github.com/DanielIoni-creator
Twitter/X @MyZubster
Dev.to dev.to/danielioni
🎯 Next Steps
Add message attachments (images, files)
Implement group chats
Add push notifications
Add message search
Built with ❤️ by Daniel Ioni and the MyZubster team.
Last updated: July 27, 2026
Top comments (0)