Backend & APIs Topic
Real-Time Chat System - System Design
1. Problem Statement
Design a scalable real-time chat system supporting:
-
One-on-one messaging
-
Group chats (up to 100,000 members)
-
Presence indicators (online/offline)
-
Message delivery receipts (sent, delivered, read)
-
Media sharing (images, videos, documents)
-
Typing indicators
-
End-to-end encryption
-
Push notifications
-
Search functionality
-
Support for 100M+ daily active users
2. System Requirements
Functional Requirements:
-
Messaging:
-
Text messages with formatting
-
Media messages (images, videos, files up to 100MB)
-
Voice messages
-
Reactions (emojis)
-
Message replies/threads
-
Edit/delete messages
-
-
Chat Features:
-
Create/join groups
-
Admin controls
-
Mute notifications
-
Pin messages
-
Starred/favorite messages
-
Chat history with infinite scroll
-
-
Real-time Features:
-
Typing indicators
-
Online presence
-
Message read receipts
-
Real-time message sync across devices
-
-
Additional Features:
-
Push notifications
-
Search messages and media
-
User profiles
-
Contact management
-
Data export
-
Non-Functional Requirements:
-
Latency: < 100ms for message delivery
-
Availability: 99.99% uptime
-
Scalability: Handle 1M+ concurrent connections
-
Durability: No message loss
-
Security: End-to-end encryption
-
Consistency: Message ordering guaranteed
-
Cost: Optimized for message volume
3. High-Level Architecture
text
┌─────────────────────────────────────────────────────────────────────┐
│ Client Applications │
├──────────────────┬──────────────────┬──────────────────┬────────────┤
│ Web (React) │ iOS (Swift) │ Android (Kotlin) │ Desktop │
└──────────────────┴──────────────────┴──────────────────┴────────────┘
│
┌─────────────────────────────▼───────────────────────────────────────┐
│ API Gateway & Load Balancer │
│ (SSL Termination, Rate Limiting, Auth) │
└──────────────────────┬───────────────────────────────────────────────┘
│
┌──────────────────────▼───────────────────────────────────────────────┐
│ Microservices Ecosystem │
├────────────┬────────────┬────────────┬────────────┬──────────────────┤
│ Chat Service│ Presence │Media Service│Notification│ Search Service│
│ │ Service │ │ Service │ │
├────────────┴────────────┴────────────┴────────────┴──────────────────┤
│ Service Mesh & Service Discovery │
└──────────────────────────────┬───────────────────────────────────────┘
│
┌──────────────────────────────▼───────────────────────────────────────┐
│ Message Queue (Kafka/Pulsar) │
│ (Event Streaming for Real-time Communication) │
└────────────────────┬─────────────────────────────────────────────────┘
│
┌────────────────▼─────────────────┐ ┌────────────────────────────┐
│ WebSocket Connection │ │ REST APIs (CRUD) │
│ Manager Cluster │ │ │
└────────────────┬─────────────────┘ └─────────────┬──────────────┘
│ │
┌────────────────▼──────────────────────────────────▼──────────────┐
│ Data Storage Layer │
├────────────┬────────────┬────────────┬────────────┬─────────────┤
│ Message DB │ User DB │ Media │ Cache │ Search │
│ (Cassandra)│ (PostgreSQL)│ (S3 + CDN) │ (Redis) │ (Elastic) │
└────────────┴────────────┴────────────┴────────────┴─────────────┘
4. Component Design Deep Dive
4.1 Real-time Communication Layer
WebSocket Server Implementation:
typescript
class WebSocketGateway {
private connections: Map<string, WebSocketConnection>;
private userToConnections: Map<string, Set<string>>;
private roomSubscriptions: Map<string, Set<string>>;
async handleConnection(socket: WebSocket, userId: string): Promise<void> {
const connectionId = this.generateConnectionId();
const connection = new WebSocketConnection(connectionId, socket, userId);
// Store connection
this.connections.set(connectionId, connection);
// Map user to multiple connections (multi-device support)
if (!this.userToConnections.has(userId)) {
this.userToConnections.set(userId, new Set());
}
this.userToConnections.get(userId)!.add(connectionId);
// Send connection acknowledgment
await this.send(connectionId, {
type: 'connected',
connectionId,
serverTime: Date.now()
});
// Start heartbeat
this.startHeartbeat(connectionId);
// Update presence
await this.presenceService.userOnline(userId);
}
async handleMessage(connectionId: string, message: any): Promise<void> {
const connection = this.connections.get(connectionId);
if (!connection) return;
switch (message.type) {
case 'ping':
await this.handlePing(connectionId);
break;
case 'subscribe':
await this.handleSubscribe(connectionId, message.roomId);
break;
case 'unsubscribe':
await this.handleUnsubscribe(connectionId, message.roomId);
break;
case 'typing':
await this.handleTypingIndicator(connectionId, message);
break;
case 'message':
await this.handleNewMessage(connectionId, message);
break;
case 'read_receipt':
await this.handleReadReceipt(connectionId, message);
break;
}
}
private async handleNewMessage(connectionId: string, message: any): Promise<void> {
const connection = this.connections.get(connectionId)!;
// Validate message
const validated = await this.validateMessage(message);
// Generate message ID
const messageId = this.generateMessageId(connection.userId);
// Store message
await this.messageService.storeMessage({
id: messageId,
senderId: connection.userId,
roomId: message.roomId,
content: message.content,
type: message.messageType || 'text',
timestamp: Date.now(),
metadata: message.metadata
});
// Publish to message queue
await this.messageQueue.publish('chat.messages', {
messageId,
roomId: message.roomId,
senderId: connection.userId,
content: message.content,
timestamp: Date.now()
});
// Send delivery receipt to sender
await this.send(connectionId, {
type: 'message_sent',
clientMessageId: message.clientMessageId,
serverMessageId: messageId,
timestamp: Date.now()
});
}
}
4.2 Message Flow Architecture
text
Message Flow: 1. Client A → WebSocket Gateway → Message Validation 2. WebSocket Gateway → Message Service → Store in DB 3. Message Service → Kafka (chat.messages topic) 4. Message Router (consumes from Kafka) ↓ 5. For each recipient: a. Check if online (Presence Service) b. If online: Route to appropriate WebSocket Gateway c. If offline: Store for later delivery 6. WebSocket Gateway → Client B 7. Client B sends read receipt 8. Update read status in DB 9. Notify sender via WebSocket
go
// Message Router in Go for high performance
type MessageRouter struct {
kafkaConsumer *kafka.Consumer
presenceService *PresenceService
wsGateways map[string]*WSGatewayClient
messageStore MessageStore
}
func (r *MessageRouter) Start() {
for {
msg, err := r.kafkaConsumer.Consume()
if err != nil {
log.Error("Failed to consume message", err)
continue
}
go r.routeMessage(msg)
}
}
func (r *MessageRouter) routeMessage(msg *Message) {
// Get room members
members, err := r.getRoomMembers(msg.RoomID)
if err != nil {
log.Error("Failed to get room members", err)
return
}
// For each member
for _, memberID := range members {
// Skip sender
if memberID == msg.SenderID {
continue
}
// Check presence
isOnline, connections := r.presenceService.GetUserConnections(memberID)
if isOnline {
// Real-time delivery
for _, conn := range connections {
gatewayClient := r.getGatewayClient(conn.GatewayID)
gatewayClient.SendMessage(conn.ConnectionID, msg)
}
// Mark as delivered
r.messageStore.MarkDelivered(msg.ID, memberID)
} else {
// Store for offline delivery
r.messageStore.StoreOfflineMessage(msg.ID, memberID)
// Queue push notification
r.notificationService.QueuePushNotification(memberID, msg)
}
}
}
4.3 Data Storage Design
Message Storage (Cassandra Schema):
cql
-- Messages table (partitioned by room_id, clustered by timestamp)
CREATE TABLE messages (
room_id uuid,
message_id timeuuid,
sender_id uuid,
content text,
message_type text,
metadata map<text, text>,
created_at timestamp,
-- Index for quick lookup
PRIMARY KEY ((room_id), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC)
AND compaction = {'class': 'TimeWindowCompactionStrategy'}
AND default_time_to_live = 31536000; -- 1 year retention
-- Offline messages table
CREATE TABLE offline_messages (
user_id uuid,
room_id uuid,
message_id timeuuid,
created_at timestamp,
PRIMARY KEY ((user_id), room_id, message_id)
);
-- Read receipts table
CREATE TABLE read_receipts (
room_id uuid,
user_id uuid,
last_read_message_id timeuuid,
updated_at timestamp,
PRIMARY KEY ((room_id), user_id)
);
-- Message reactions
CREATE TABLE message_reactions (
message_id timeuuid,
user_id uuid,
reaction text,
created_at timestamp,
PRIMARY KEY ((message_id), user_id)
);
User Data (PostgreSQL):
sql
-- Users table
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
email VARCHAR(255) UNIQUE NOT NULL,
phone VARCHAR(20) UNIQUE,
username VARCHAR(50) UNIQUE NOT NULL,
full_name VARCHAR(255),
avatar_url TEXT,
status VARCHAR(50) DEFAULT 'active',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-- JSONB for flexible metadata
metadata JSONB DEFAULT '{}'
);
-- Contacts/relationships
CREATE TABLE contacts (
user_id UUID REFERENCES users(id),
contact_id UUID REFERENCES users(id),
relationship VARCHAR(50) DEFAULT 'friend', -- friend, blocked, pending
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, contact_id)
);
-- Chat rooms
CREATE TABLE rooms (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(255),
type VARCHAR(20) DEFAULT 'direct', -- direct, group, channel
created_by UUID REFERENCES users(id),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata JSONB DEFAULT '{}'
);
-- Room members with roles
CREATE TABLE room_members (
room_id UUID REFERENCES rooms(id),
user_id UUID REFERENCES users(id),
role VARCHAR(20) DEFAULT 'member', -- member, admin, owner
joined_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_read_message_id UUID,
muted_until TIMESTAMP,
PRIMARY KEY (room_id, user_id)
);
Redis for Real-time State:
python
class PresenceService:
def __init__(self, redis_client):
self.redis = redis_client
self.online_users_key = "online_users"
self.user_connections_key = "user:connections:{user_id}"
self.user_presence_key = "user:presence:{user_id}"
async def user_online(self, user_id: str, connection_id: str, device_info: dict):
# Add to online users set
await self.redis.sadd(self.online_users_key, user_id)
# Store connection info
connections_key = self.user_connections_key.format(user_id=user_id)
connection_data = {
'connection_id': connection_id,
'device_info': json.dumps(device_info),
'last_seen': time.time()
}
await self.redis.hset(connections_key, connection_id, json.dumps(connection_data))
# Set presence status with expiry
presence_key = self.user_presence_key.format(user_id=user_id)
await self.redis.setex(presence_key, 60, "online") # Refresh every 60 seconds
# Publish presence update
await self.redis.publish(f"presence:{user_id}", json.dumps({
'user_id': user_id,
'status': 'online',
'timestamp': time.time()
}))
async def user_offline(self, user_id: str, connection_id: str):
# Remove connection
connections_key = self.user_connections_key.format(user_id=user_id)
await self.redis.hdel(connections_key, connection_id)
# Check if user has any remaining connections
remaining_connections = await self.redis.hlen(connections_key)
if remaining_connections == 0:
# Remove from online users set
await self.redis.srem(self.online_users_key, user_id)
# Update presence status
presence_key = self.user_presence_key.format(user_id=user_id)
await self.redis.setex(presence_key, 300, "offline") # Keep for 5 minutes
# Publish presence update
await self.redis.publish(f"presence:{user_id}", json.dumps({
'user_id': user_id,
'status': 'offline',
'timestamp': time.time()
}))
4.4 Message Queue Architecture
java
// Kafka topics configuration
public class KafkaConfig {
@Bean
public NewTopic chatMessagesTopic() {
return TopicBuilder.name("chat.messages")
.partitions(100) // Partition by room_id hash
.replicas(3)
.config(TopicConfig.RETENTION_MS_CONFIG, "604800000") // 7 days
.build();
}
@Bean
public NewTopic typingIndicatorsTopic() {
return TopicBuilder.name("chat.typing")
.partitions(50)
.replicas(2)
.config(TopicConfig.RETENTION_MS_CONFIG, "86400000") // 1 day
.build();
}
@Bean
public NewTopic presenceEventsTopic() {
return TopicBuilder.name("presence.events")
.partitions(30)
.replicas(2)
.config(TopicConfig.RETENTION_MS_CONFIG, "86400000")
.build();
}
@Bean
public NewTopic notificationEventsTopic() {
return TopicBuilder.name("notifications.events")
.partitions(50)
.replicas(2)
.build();
}
}
// Message producer with partitioning
@Component
public class MessageProducer {
private final KafkaTemplate<String, ChatMessage> kafkaTemplate;
public void sendMessage(ChatMessage message) {
// Partition by room_id for message ordering guarantee
int partition = Math.abs(message.getRoomId().hashCode()) % 100;
kafkaTemplate.send(
"chat.messages",
partition,
message.getRoomId(),
message
).addCallback(
result -> log.info("Message sent successfully"),
ex -> log.error("Failed to send message", ex)
);
}
public void sendTypingIndicator(String roomId, String userId, boolean isTyping) {
TypingEvent event = new TypingEvent(roomId, userId, isTyping, System.currentTimeMillis());
// Use consistent hashing for same user typing events
String key = roomId + ":" + userId;
kafkaTemplate.send("chat.typing", key, event);
}
}
4.5 Media Handling Service
typescript
class MediaService {
private s3Client: AWS.S3;
private imageProcessor: Sharp;
private videoProcessor: FFmpeg;
async uploadMedia(
file: Buffer,
userId: string,
metadata: MediaMetadata
): Promise<MediaResponse> {
// Generate unique media ID
const mediaId = crypto.randomUUID();
// Determine storage path
const path = `media/${userId}/${Date.now()}-${mediaId}`;
// Process based on type
let processedFile: Buffer;
let thumbnail: Buffer | null = null;
switch (metadata.type) {
case 'image':
processedFile = await this.processImage(file, metadata);
thumbnail = await this.generateThumbnail(processedFile);
break;
case 'video':
processedFile = await this.processVideo(file, metadata);
thumbnail = await this.extractVideoThumbnail(file);
break;
case 'audio':
processedFile = await this.processAudio(file, metadata);
break;
default:
processedFile = file;
}
// Upload to S3
const uploadParams = {
Bucket: process.env.MEDIA_BUCKET,
Key: path,
Body: processedFile,
ContentType: metadata.mimeType,
Metadata: {
'user-id': userId,
'media-id': mediaId,
'original-filename': metadata.filename
}
};
await this.s3Client.putObject(uploadParams);
// Upload thumbnail if exists
let thumbnailUrl: string | null = null;
if (thumbnail) {
const thumbnailPath = `thumbnails/${userId}/${mediaId}.jpg`;
await this.s3Client.putObject({
Bucket: process.env.MEDIA_BUCKET,
Key: thumbnailPath,
Body: thumbnail,
ContentType: 'image/jpeg'
});
thumbnailUrl = await this.getCDNUrl(thumbnailPath);
}
// Store metadata in database
await this.storeMediaMetadata({
id: mediaId,
userId,
path,
type: metadata.type,
mimeType: metadata.mimeType,
size: processedFile.length,
dimensions: metadata.dimensions,
duration: metadata.duration,
thumbnailUrl,
createdAt: new Date()
});
return {
mediaId,
url: await this.getCDNUrl(path),
thumbnailUrl,
size: processedFile.length
};
}
private async processImage(file: Buffer, metadata: MediaMetadata): Promise<Buffer> {
const image = this.imageProcessor(file);
// Resize if too large
if (metadata.dimensions &&
(metadata.dimensions.width > 1920 || metadata.dimensions.height > 1920)) {
image.resize(1920, 1920, { fit: 'inside' });
}
// Compress
image.jpeg({ quality: 85 });
return await image.toBuffer();
}
private async generateThumbnail(image: Buffer): Promise<Buffer> {
return await this.imageProcessor(image)
.resize(300, 300, { fit: 'cover' })
.jpeg({ quality: 70 })
.toBuffer();
}
}
5. Scalability Patterns
5.1 WebSocket Connection Scaling
go
// WebSocket Gateway with consistent hashing
type GatewayManager struct {
gateways map[string]*GatewayInfo
hashRing *consistenthash.Map
redisClient *redis.Client
}
func (gm *GatewayManager) RegisterGateway(gatewayID string, gatewayInfo GatewayInfo) {
gm.gateways[gatewayID] = &gatewayInfo
// Add to hash ring with weight based on capacity
gm.hashRing.Add(gatewayID)
// Store in Redis for service discovery
gm.redisClient.HSet("chat:gateways", gatewayID, gatewayInfo.ToJSON())
}
func (gm *GatewayManager) GetGatewayForUser(userID string) (string, error) {
// Consistent hashing for user-to-gateway mapping
gatewayID, ok := gm.hashRing.Get(userID)
if !ok {
return "", errors.New("no gateway available")
}
return gatewayID, nil
}
func (gm *GatewayManager) RouteMessage(userID string, message Message) error {
gatewayID, err := gm.GetGatewayForUser(userID)
if err != nil {
return err
}
gateway, exists := gm.gateways[gatewayID]
if !exists {
return errors.New("gateway not found")
}
// Forward message to appropriate gateway
return gm.sendToGateway(gateway, userID, message)
}
5.2 Database Sharding Strategy
sql
-- Sharding by user_id for messages
-- Using Citus (PostgreSQL extension) for distributed tables
-- Enable Citus extension
CREATE EXTENSION citus;
-- Create distributed tables
SELECT create_distributed_table('messages', 'room_id');
SELECT create_distributed_table('users', 'id');
SELECT create_distributed_table('room_members', 'room_id');
-- Configure shard count based on data size
SET citus.shard_count = 64; -- For messages table
SET citus.shard_replication_factor = 2;
-- Create reference tables (small, replicated to all nodes)
SELECT create_reference_table('rooms');
-- Query optimization with colocation
-- Rooms and messages are colocated by room_id
SELECT mark_tables_colocated('rooms', 'messages');
5.3 Caching Strategy
python
class ChatCache:
def __init__(self):
self.redis = RedisCluster()
self.local_cache = {} # In-memory cache for hot data
async def get_user_presence(self, user_id: str) -> Optional[str]:
# Check local cache first
if user_id in self.local_cache.get('presence', {}):
return self.local_cache['presence'][user_id]
# Check Redis
presence = await self.redis.get(f"presence:{user_id}")
if presence:
# Update local cache with TTL
self.local_cache.setdefault('presence', {})[user_id] = presence
# Schedule local cache eviction
asyncio.create_task(self.evict_from_local_cache('presence', user_id, 30))
return presence
async def get_room_members(self, room_id: str) -> List[str]:
cache_key = f"room:{room_id}:members"
# Try to get from Redis
members = await self.redis.get(cache_key)
if members:
return json.loads(members)
# Get from database
members = await self.db.get_room_members(room_id)
# Store in Redis with expiration
await self.redis.setex(cache_key, 300, json.dumps(members))
return members
async def get_message_history(self, room_id: str, limit: int = 50, offset: int = 0):
cache_key = f"room:{room_id}:messages:{limit}:{offset}"
# Use Redis sorted set for message history caching
cached = await self.redis.zrevrange(cache_key, 0, -1)
if cached and len(cached) >= limit:
return [json.loads(msg) for msg in cached]
# Get from database
messages = await self.db.get_messages(room_id, limit, offset)
# Cache in Redis with TTL
pipeline = self.redis.pipeline()
for msg in messages:
pipeline.zadd(cache_key, {json.dumps(msg): msg['timestamp']})
pipeline.expire(cache_key, 600) # 10 minutes
await pipeline.execute()
return messages
6. Real-time Features Implementation
6.1 Typing Indicators
typescript
class TypingIndicatorService {
private typingSessions: Map<string, TypingSession> = new Map();
async handleTypingStart(userId: string, roomId: string): Promise<void> {
const sessionKey = `${userId}:${roomId}`;
// Check if already typing
if (this.typingSessions.has(sessionKey)) {
return;
}
// Create new typing session
const session: TypingSession = {
userId,
roomId,
startedAt: Date.now(),
lastActionAt: Date.now()
};
this.typingSessions.set(sessionKey, session);
// Notify room members
await this.notifyRoom(roomId, {
type: 'typing_start',
userId,
roomId,
timestamp: Date.now()
});
// Schedule cleanup if user stops typing
setTimeout(() => {
this.checkAndCleanupSession(sessionKey);
}, 3000);
}
async handleTypingStop(userId: string, roomId: string): Promise<void> {
const sessionKey = `${userId}:${roomId}`;
if (this.typingSessions.has(sessionKey)) {
this.typingSessions.delete(sessionKey);
// Notify room members
await this.notifyRoom(roomId, {
type: 'typing_stop',
userId,
roomId,
timestamp: Date.now()
});
}
}
private checkAndCleanupSession(sessionKey: string): void {
const session = this.typingSessions.get(sessionKey);
if (!session) return;
// If no action for 3 seconds, assume stopped typing
if (Date.now() - session.lastActionAt > 3000) {
this.typingSessions.delete(sessionKey);
this.notifyRoom(session.roomId, {
type: 'typing_stop',
userId: session.userId,
roomId: session.roomId,
timestamp: Date.now()
});
}
}
}
6.2 Read Receipts
java
public class ReadReceiptService {
private final MessageRepository messageRepository;
private final NotificationService notificationService;
@Transactional
public void markMessagesAsRead(String userId, String roomId, String lastReadMessageId) {
// Update last read message for user in this room
messageRepository.updateLastReadMessage(userId, roomId, lastReadMessageId);
// Get messages that are now read
List<Message> newlyReadMessages = messageRepository
.findMessagesBefore(userId, roomId, lastReadMessageId);
// Update read status for each message
for (Message message : newlyReadMessages) {
messageRepository.addReadReceipt(
message.getId(),
userId,
Instant.now()
);
}
// Notify sender(s) that their messages were read
Set<String> senders = newlyReadMessages.stream()
.map(Message::getSenderId)
.collect(Collectors.toSet());
for (String senderId : senders) {
if (!senderId.equals(userId)) {
notificationService.notifyMessageRead(
senderId,
roomId,
lastReadMessageId,
userId
);
}
}
}
public MessageReadStatus getReadStatus(String messageId) {
List<ReadReceipt> receipts = messageRepository
.findReadReceipts(messageId);
Set<String> readBy = receipts.stream()
.map(ReadReceipt::getUserId)
.collect(Collectors.toSet());
// Get room members
Message message = messageRepository.findById(messageId);
List<String> roomMembers = messageRepository
.getRoomMembers(message.getRoomId());
// Calculate delivery status
Set<String> deliveredTo = new HashSet<>();
Set<String> pendingDelivery = new HashSet<>();
for (String memberId : roomMembers) {
if (memberId.equals(message.getSenderId())) {
continue;
}
// Check if user was in room when message was sent
if (wasUserInRoom(memberId, message.getRoomId(), message.getTimestamp())) {
if (readBy.contains(memberId)) {
// Message was read
} else if (isUserOnline(memberId)) {
// Message was delivered but not read
deliveredTo.add(memberId);
} else {
// Message not yet delivered
pendingDelivery.add(memberId);
}
}
}
return new MessageReadStatus(
messageId,
readBy,
deliveredTo,
pendingDelivery
);
}
}
7. Security & Encryption
7.1 End-to-End Encryption
typescript
class EndToEndEncryption {
private readonly crypto = new WebCrypto();
async generateKeyPair(): Promise<CryptoKeyPair> {
return await this.crypto.subtle.generateKey(
{
name: "ECDH",
namedCurve: "P-256"
},
true, // extractable
["deriveKey"]
);
}
async encryptMessage(
message: string,
senderPrivateKey: CryptoKey,
recipientPublicKey: CryptoKey
): Promise<EncryptedMessage> {
// Generate ephemeral key pair for forward secrecy
const ephemeralKeyPair = await this.generateKeyPair();
// Derive shared secret
const sharedSecret = await this.crypto.subtle.deriveKey(
{
name: "ECDH",
public: recipientPublicKey
},
senderPrivateKey,
{
name: "AES-GCM",
length: 256
},
true,
["encrypt"]
);
// Generate random IV
const iv = this.crypto.getRandomValues(new Uint8Array(12));
// Encrypt message
const encryptedContent = await this.crypto.subtle.encrypt(
{
name: "AES-GCM",
iv: iv
},
sharedSecret,
new TextEncoder().encode(message)
);
return {
ciphertext: this.arrayBufferToBase64(encryptedContent),
ephemeralPublicKey: await this.exportPublicKey(ephemeralKeyPair.publicKey),
iv: this.arrayBufferToBase64(iv),
signature: await this.signMessage(message, senderPrivateKey)
};
}
async decryptMessage(
encrypted: EncryptedMessage,
recipientPrivateKey: CryptoKey,
senderPublicKey: CryptoKey
): Promise<string> {
// Import ephemeral public key
const ephemeralPublicKey = await this.importPublicKey(encrypted.ephemeralPublicKey);
// Derive shared secret
const sharedSecret = await this.crypto.subtle.deriveKey(
{
name: "ECDH",
public: ephemeralPublicKey
},
recipientPrivateKey,
{
name: "AES-GCM",
length: 256
},
true,
["decrypt"]
);
// Decrypt message
const decrypted = await this.crypto.subtle.decrypt(
{
name: "AES-GCM",
iv: this.base64ToArrayBuffer(encrypted.iv)
},
sharedSecret,
this.base64ToArrayBuffer(encrypted.ciphertext)
);
const message = new TextDecoder().decode(decrypted);
// Verify signature
const isValid = await this.verifySignature(
message,
encrypted.signature,
senderPublicKey
);
if (!isValid) {
throw new Error("Invalid message signature");
}
return message;
}
}
7.2 Rate Limiting & Abuse Prevention
go
type RateLimiter struct {
redisClient *redis.Client
limits map[string]RateLimitConfig
}
type RateLimitConfig struct {
Requests int
Window time.Duration
Burst int
}
func (rl *RateLimiter) CheckLimit(userID, action string) (bool, error) {
key := fmt.Sprintf("ratelimit:%s:%s", userID, action)
config, exists := rl.limits[action]
if !exists {
return true, nil // No limit configured
}
// Use Redis sorted set for sliding window
now := time.Now().UnixMilli()
windowStart := now - config.Window.Milliseconds()
// Remove old entries
rl.redisClient.ZRemRangeByScore(key, "0", strconv.FormatInt(windowStart, 10))
// Count current requests
count, err := rl.redisClient.ZCard(key).Result()
if err != nil {
return false, err
}
if int(count) >= config.Requests {
// Check burst allowance
burstKey := fmt.Sprintf("burst:%s:%s", userID, action)
burstCount, _ := rl.redisClient.Get(burstKey).Int()
if burstCount >= config.Burst {
return false, nil
}
// Allow burst
rl.redisClient.Incr(burstKey)
rl.redisClient.Expire(burstKey, config.Window)
}
// Add new request
rl.redisClient.ZAdd(key, redis.Z{
Score: float64(now),
Member: strconv.FormatInt(now, 10),
})
rl.redisClient.Expire(key, config.Window)
return true, nil
}
// Configure limits
limiter := &RateLimiter{
limits: map[string]RateLimitConfig{
"send_message": {Requests: 60, Window: time.Minute, Burst: 10},
"create_room": {Requests: 10, Window: time.Hour, Burst: 3},
"add_member": {Requests: 100, Window: time.Hour, Burst: 20},
"typing": {Requests: 30, Window: time.Minute, Burst: 5},
},
}
8. Search & Indexing
python
class MessageSearchService:
def __init__(self, elasticsearch_client):
self.es = elasticsearch_client
self.index_name = "chat_messages"
async def index_message(self, message: dict):
# Create document for indexing
doc = {
"message_id": message["id"],
"room_id": message["room_id"],
"sender_id": message["sender_id"],
"content": message["content"],
"content_type": message.get("type", "text"),
"timestamp": message["timestamp"],
"attachments": message.get("attachments", []),
"mentions": self.extract_mentions(message["content"])
}
# Index in Elasticsearch
await self.es.index(
index=self.index_name,
id=message["id"],
document=doc,
refresh="wait_for"
)
async def search_messages(
self,
user_id: str,
query: str,
filters: dict,
page: int = 1,
size: int = 20
):
# Get user's accessible rooms
accessible_rooms = await self.get_accessible_rooms(user_id)
# Build Elasticsearch query
search_body = {
"query": {
"bool": {
"must": [
{
"simple_query_string": {
"query": query,
"fields": ["content"],
"default_operator": "AND"
}
},
{
"terms": {
"room_id": accessible_rooms
}
}
],
"filter": self.build_filters(filters)
}
},
"sort": [
{"timestamp": {"order": "desc"}}
],
"from": (page - 1) * size,
"size": size,
"highlight": {
"fields": {
"content": {
"number_of_fragments": 3,
"fragment_size": 150
}
}
}
}
response = await self.es.search(
index=self.index_name,
body=search_body
)
# Process results
results = []
for hit in response["hits"]["hits"]:
result = {
"message_id": hit["_source"]["message_id"],
"room_id": hit["_source"]["room_id"],
"sender_id": hit["_source"]["sender_id"],
"content": hit["_source"]["content"],
"timestamp": hit["_source"]["timestamp"],
"score": hit["_score"],
"highlight": hit.get("highlight", {})
}
results.append(result)
return {
"results": results,
"total": response["hits"]["total"]["value"],
"page": page,
"total_pages": math.ceil(response["hits"]["total"]["value"] / size)
}
def build_filters(self, filters: dict) -> list:
filter_list = []
if filters.get("room_id"):
filter_list.append({"term": {"room_id": filters["room_id"]}})
if filters.get("sender_id"):
filter_list.append({"term": {"sender_id": filters["sender_id"]}})
if filters.get("date_from") and filters.get("date_to"):
filter_list.append({
"range": {
"timestamp": {
"gte": filters["date_from"],
"lte": filters["date_to"]
}
}
})
if filters.get("has_attachment"):
filter_list.append({"exists": {"field": "attachments"}})
if filters.get("mentioned_user"):
filter_list.append({"term": {"mentions": filters["mentioned_user"]}})
return filter_list
9. Push Notifications
typescript
class PushNotificationService {
private apns: apns.Provider;
private fcm: admin.messaging.Messaging;
private webPush: webpush.WebPush;
async sendNotification(
userId: string,
notification: NotificationPayload
): Promise<void> {
// Get user's devices
const devices = await this.getUserDevices(userId);
// Group devices by platform
const groupedDevices = this.groupDevicesByPlatform(devices);
// Send to each platform
const promises = [];
// iOS (APNs)
if (groupedDevices.ios.length > 0) {
promises.push(this.sendAPNS(groupedDevices.ios, notification));
}
// Android (FCM)
if (groupedDevices.android.length > 0) {
promises.push(this.sendFCM(groupedDevices.android, notification));
}
// Web (Web Push)
if (groupedDevices.web.length > 0) {
promises.push(this.sendWebPush(groupedDevices.web, notification));
}
await Promise.all(promises);
}
private async sendAPNS(deviceTokens: string[], notification: NotificationPayload) {
const notifications = deviceTokens.map(token => ({
deviceToken: token,
body: {
aps: {
alert: {
title: notification.title,
body: notification.body,
},
badge: notification.badge || 1,
sound: "default",
"thread-id": notification.threadId,
"mutable-content": 1
},
data: notification.data
}
}));
return this.apns.sendMany(notifications);
}
private async sendFCM(deviceTokens: string[], notification: NotificationPayload) {
const message: admin.messaging.MulticastMessage = {
tokens: deviceTokens,
notification: {
title: notification.title,
body: notification.body,
imageUrl: notification.imageUrl
},
data: notification.data,
android: {
priority: "high",
notification: {
channelId: "chat_messages",
icon: "ic_notification",
color: "#2196F3",
tag: notification.threadId,
clickAction: "OPEN_CHAT"
}
},
apns: {
payload: {
aps: {
badge: notification.badge || 1,
sound: "default"
}
}
}
};
return this.fcm.sendMulticast(message);
}
private async sendWebPush(subscriptions: PushSubscription[], notification: NotificationPayload) {
const payload = JSON.stringify({
title: notification.title,
body: notification.body,
icon: notification.icon || "/icon-192.png",
badge: notification.badgeIcon || "/badge-72.png",
tag: notification.threadId,
data: notification.data,
actions: [
{
action: "reply",
title: "Reply",
icon: "/reply-icon.png"
},
{
action: "mark_read",
title: "Mark as Read"
}
],
vibrate: [200, 100, 200],
timestamp: Date.now(),
requireInteraction: true
});
const promises = subscriptions.map(async subscription => {
try {
await this.webPush.sendNotification(subscription, payload);
} catch (error) {
// If subscription is invalid, remove it
if (error.statusCode === 410) {
await this.removeInvalidSubscription(subscription);
}
}
});
await Promise.all(promises);
}
}
10. Monitoring & Analytics
python
class ChatAnalytics:
def __init__(self):
self.statsd_client = statsd.StatsClient()
self.prometheus_registry = CollectorRegistry()
# Define metrics
self.messages_sent = Counter(
'chat_messages_sent_total',
'Total messages sent',
['room_type', 'message_type'],
registry=self.prometheus_registry
)
self.message_delivery_latency = Histogram(
'chat_message_delivery_latency_seconds',
'Message delivery latency',
['room_type'],
buckets=[0.01, 0.05, 0.1, 0.5, 1, 5],
registry=self.prometheus_registry
)
self.active_connections = Gauge(
'chat_active_connections',
'Number of active WebSocket connections',
registry=self.prometheus_registry
)
async def track_message_sent(self, message: dict):
# Increment counter
self.messages_sent.labels(
room_type=message.get('room_type', 'direct'),
message_type=message.get('type', 'text')
).inc()
# Track to StatsD
self.statsd_client.incr('messages.sent', tags=[
f"room_type:{message.get('room_type', 'direct')}",
f"message_type:{message.get('type', 'text')}"
])
async def track_message_delivery(self, message_id: str, sent_time: int, delivered_time: int):
latency = (delivered_time - sent_time) / 1000.0 # Convert to seconds
self.message_delivery_latency.labels(
room_type=self.get_room_type(message_id)
).observe(latency)
# Log to Elasticsearch for detailed analysis
await self.es.index(
index='chat_delivery_metrics',
body={
'message_id': message_id,
'sent_time': sent_time,
'delivered_time': delivered_time,
'latency': latency,
'@timestamp': datetime.utcnow()
}
)
async def generate_daily_report(self, date: datetime):
# Get message volume by hour
message_volume = await self.get_message_volume_by_hour(date)
# Get active users
active_users = await self.get_active_users(date)
# Get delivery success rate
delivery_rate = await self.get_delivery_success_rate(date)
# Get top rooms by activity
top_rooms = await self.get_top_rooms(date, limit=10)
# Generate report
report = {
'date': date.isoformat(),
'total_messages': sum(hour['count'] for hour in message_volume),
'unique_active_users': len(active_users),
'delivery_success_rate': delivery_rate,
'peak_hour': max(message_volume, key=lambda x: x['count']),
'top_rooms': top_rooms,
'message_types': await self.get_message_type_distribution(date),
'platform_distribution': await self.get_platform_distribution(date)
}
# Store report
await self.es.index(
index='chat_daily_reports',
id=date.strftime('%Y-%m-%d'),
body=report
)
return report
11. Testing Strategy
typescript
describe('Chat System', () => {
describe('WebSocket Connection', () => {
it('should establish connection successfully', async () => {
const ws = new WebSocket('ws://localhost:8080/chat');
await new Promise((resolve) => {
ws.onopen = resolve;
});
expect(ws.readyState).toBe(WebSocket.OPEN);
});
it('should handle multiple connections per user', async () => {
// Simulate user connecting from multiple devices
const connections = await Promise.all([
connectUser('user1'),
connectUser('user1'),
connectUser('user1')
]);
// All connections should be active
connections.forEach(conn => {
expect(conn.status).toBe('connected');
});
// Presence service should show user as online
const presence = await presenceService.getPresence('user1');
expect(presence).toBe('online');
});
});
describe('Message Delivery', () => {
it('should deliver messages in order', async () => {
const [user1, user2] = await setupTestUsers();
// Send 100 messages quickly
const messages = [];
for (let i = 0; i < 100; i++) {
messages.push(user1.sendMessage(user2.id, `Message ${i}`));
}
await Promise.all(messages);
// Verify all messages received in order
const received = user2.getReceivedMessages();
expect(received.length).toBe(100);
for (let i = 0; i < 100; i++) {
expect(received[i].content).toBe(`Message ${i}`);
}
});
it('should handle delivery to offline users', async () => {
const [user1, user2] = await setupTestUsers();
// User2 goes offline
await user2.disconnect();
// Send message
await user1.sendMessage(user2.id, 'Offline message');
// User2 comes back online
await user2.connect();
// Should receive pending message
await waitFor(() => {
expect(user2.getReceivedMessages().length).toBe(1);
});
});
});
describe('Performance Tests', () => {
it('should handle 10,000 concurrent connections', async () => {
const connections = [];
// Create 10,000 connections
for (let i = 0; i < 10000; i++) {
connections.push(createConnection(`user${i}`));
}
await Promise.all(connections);
// Verify all connected
const connected = connections.filter(c => c.isConnected());
expect(connected.length).toBe(10000);
}, 30000);
it('should process 1,000 messages per second', async () => {
const startTime = Date.now();
const messageCount = 5000;
// Send messages in parallel
const promises = [];
for (let i = 0; i < messageCount; i++) {
promises.push(sendTestMessage(i));
}
await Promise.all(promises);
const endTime = Date.now();
const duration = (endTime - startTime) / 1000; // seconds
const messagesPerSecond = messageCount / duration;
expect(messagesPerSecond).toBeGreaterThan(1000);
});
});
});
12. Deployment & DevOps
yaml
# Kubernetes deployment configuration
apiVersion: apps/v1
kind: Deployment
metadata:
name: websocket-gateway
spec:
replicas: 10
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 2
maxUnavailable: 1
selector:
matchLabels:
app: websocket-gateway
template:
metadata:
labels:
app: websocket-gateway
spec:
containers:
- name: gateway
image: chat-gateway:latest
ports:
- containerPort: 8080
env:
- name: NODE_ENV
value: production
- name: REDIS_HOST
value: redis-cluster
- name: KAFKA_BROKERS
value: kafka-0.kafka:9092,kafka-1.kafka:9092
resources:
requests:
memory: "512Mi"
cpu: "250m"
limits:
memory: "1Gi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
periodSeconds: 5
---
# Horizontal Pod Autoscaler
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: websocket-gateway-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: websocket-gateway
minReplicas: 5
maxReplicas: 50
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
- type: Pods
pods:
metric:
name: connections_per_pod
target:
type: AverageValue
averageValue: 10000
Key Challenges & Solutions:
-
Connection State Management:
-
Use Redis for distributed connection tracking
-
Implement heartbeat mechanism
-
Graceful reconnection with session recovery
-
-
Message Ordering:
-
Use sequence numbers or timestamps
-
Partition Kafka topics by room_id
-
Client-side buffering and reordering
-
-
Offline Support:
-
Store undelivered messages
-
Sync on reconnection
-
Conflict resolution for concurrent edits
-
-
Scale to Millions:
-
Shard by user_id or room_id
-
Use connection pooling
-
Implement rate limiting
-
-
Data Privacy:
-
End-to-end encryption
-
Ephemeral messages
-
Data retention policies
-
-
Cost Optimization:
-
Compress messages
-
Use efficient protocols (Protobuf over JSON)
-
Implement message batching
-
Use spot instances for non-critical services
-
Advanced Features Roadmap:
-
Voice/Video Calls: WebRTC integration
-
Screen Sharing: SDP/ICE protocol implementation
-
Bots & Automation: Plugin architecture
-
Message Translation: Real-time translation service
-
Reactions & Polls: Rich interactive messages
-
Message Scheduling: Send later functionality
-
Advanced Search: Semantic search, image search
-
Chat Bots: AI-powered assistants
-
Marketplace Integration: E-commerce within chats
-
AR/VR Chat: 3D spatial audio/video
This system design provides a robust, scalable foundation for a real-time chat application that can handle millions of users while maintaining low latency, high availability, and strong security.
Comments (0)
No comments yet. Be the first to comment!