563 lines
19 KiB
TypeScript
563 lines
19 KiB
TypeScript
import {
|
|
WebSocketGateway,
|
|
WebSocketServer,
|
|
SubscribeMessage,
|
|
OnGatewayConnection,
|
|
OnGatewayDisconnect,
|
|
ConnectedSocket,
|
|
MessageBody,
|
|
} from '@nestjs/websockets';
|
|
import { Server, Socket } from 'socket.io';
|
|
import { JwtService } from '@nestjs/jwt';
|
|
import { ConfigService } from '@nestjs/config';
|
|
import { MessagesService } from './messages.service';
|
|
import { SupportChatService } from '../support-chat/support-chat.service';
|
|
import { ConnectionRequestsService } from '../connection-requests/connection-requests.service';
|
|
import { PrismaService } from '../prisma/prisma.service';
|
|
import { CreateMessageDto } from './dto';
|
|
import { Logger } from '@nestjs/common';
|
|
import { RedisPresenceService } from '../common/services/redis-presence.service';
|
|
|
|
interface AuthenticatedSocket extends Socket {
|
|
userId?: string;
|
|
userRole?: string;
|
|
}
|
|
|
|
@WebSocketGateway({
|
|
cors: {
|
|
origin: '*',
|
|
credentials: true,
|
|
},
|
|
transports: ['websocket', 'polling'],
|
|
// Detect dead connections quickly when a client drops network abruptly
|
|
// (e.g. airplane mode). Default is 25s ping + 20s timeout = ~45s to notice.
|
|
// 5s ping + 5s timeout = ~10s detection, which keeps presence/ticks honest.
|
|
pingInterval: 5000,
|
|
pingTimeout: 5000,
|
|
})
|
|
export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect {
|
|
@WebSocketServer()
|
|
server: Server;
|
|
|
|
private readonly logger = new Logger(MessagesGateway.name);
|
|
private userSocketMap = new Map<string, Set<string>>(); // userId -> Set of socketIds
|
|
private rateLimitMap = new Map<string, { typing: number; message: number }>(); // userId -> last event timestamps
|
|
|
|
constructor(
|
|
private readonly messagesService: MessagesService,
|
|
private readonly supportChatService: SupportChatService,
|
|
private readonly connectionRequestsService: ConnectionRequestsService,
|
|
private readonly prisma: PrismaService,
|
|
private readonly jwtService: JwtService,
|
|
private readonly configService: ConfigService,
|
|
private readonly redisPresence: RedisPresenceService,
|
|
) {
|
|
// Periodic cleanup of userSocketMap and rateLimitMap every 5 minutes
|
|
setInterval(() => {
|
|
this.logger.log(`Socket map size: ${this.userSocketMap.size} users`);
|
|
this.rateLimitMap.clear();
|
|
}, 5 * 60 * 1000);
|
|
}
|
|
|
|
/**
|
|
* Simple rate limiter - returns true if action is allowed
|
|
*/
|
|
private checkRateLimit(userId: string, action: 'typing' | 'message'): boolean {
|
|
const now = Date.now();
|
|
if (!this.rateLimitMap.has(userId)) {
|
|
this.rateLimitMap.set(userId, { typing: 0, message: 0 });
|
|
}
|
|
const limits = this.rateLimitMap.get(userId)!;
|
|
const minInterval = action === 'typing' ? 2000 : 1000; // 2s for typing, 1s for messages
|
|
if (now - limits[action] < minInterval) {
|
|
return false;
|
|
}
|
|
limits[action] = now;
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Handle new WebSocket connections
|
|
*/
|
|
async handleConnection(client: AuthenticatedSocket) {
|
|
this.logger.log(`New connection attempt: ${client.id}`);
|
|
|
|
// Extract token from handshake
|
|
const token =
|
|
client.handshake.auth?.token ||
|
|
client.handshake.headers?.authorization?.replace('Bearer ', '');
|
|
|
|
if (!token) {
|
|
this.logger.warn(`Connection rejected: No token provided for ${client.id}`);
|
|
client.disconnect(true);
|
|
return;
|
|
}
|
|
|
|
// Verify JWT token
|
|
let payload;
|
|
try {
|
|
payload = await this.jwtService.verifyAsync(token, {
|
|
secret: this.configService.get<string>('JWT_SECRET'),
|
|
});
|
|
this.logger.log(`Token verified for ${client.id}, user: ${payload.sub}`);
|
|
} catch (jwtError) {
|
|
this.logger.warn(`Connection rejected: Invalid token for ${client.id} - ${jwtError.message}`);
|
|
client.disconnect(true);
|
|
return;
|
|
}
|
|
|
|
// Attach user info to socket
|
|
const userId = payload.sub as string;
|
|
client.userId = userId;
|
|
client.userRole = payload.role;
|
|
|
|
// Track socket connection
|
|
if (!this.userSocketMap.has(userId)) {
|
|
this.userSocketMap.set(userId, new Set());
|
|
}
|
|
this.userSocketMap.get(userId)!.add(client.id);
|
|
|
|
this.logger.log(`Client connected successfully: ${client.id} (User: ${userId}, Role: ${payload.role})`);
|
|
|
|
// Join per-user room for Redis-backed cross-instance message routing
|
|
client.join(`user:${userId}`);
|
|
|
|
// Everything below is best-effort — don't let DB errors kill the socket connection
|
|
try {
|
|
// Auto-join admin users to admin_room for support chat notifications
|
|
if (payload.role === 'ADMIN' || payload.role === 'SUPER_ADMIN') {
|
|
client.join('admin_room');
|
|
this.logger.log(`Admin ${userId} joined admin_room`);
|
|
}
|
|
} catch (err) {
|
|
this.logger.warn(`Failed to join admin room for ${userId}: ${err.message}`);
|
|
}
|
|
|
|
// Update user online status in Redis + DB (don't await)
|
|
this.redisPresence.setOnline(userId).catch((err) => {
|
|
this.logger.warn(`Failed to set Redis presence for ${userId}: ${err.message}`);
|
|
});
|
|
|
|
this.messagesService.updateOnlineStatus(userId, true)
|
|
.then(() => {
|
|
this.broadcastUserStatus(userId, true);
|
|
this.logger.log(`Online status updated for ${userId}`);
|
|
})
|
|
.catch((err) => {
|
|
this.logger.warn(`Failed to update online status for ${userId}: ${err.message}`);
|
|
});
|
|
|
|
// Mark pending SENT messages as DELIVERED now that this user is online
|
|
this.messagesService.markPendingMessagesAsDelivered(userId)
|
|
.then((results) => {
|
|
for (const result of results) {
|
|
const deliveredEvent = {
|
|
messageIds: result.messageIds,
|
|
conversationId: result.conversationId,
|
|
deliveredAt: result.deliveredAt.toISOString(),
|
|
};
|
|
this.sendToUser(result.senderId, 'message_delivered', deliveredEvent);
|
|
}
|
|
if (results.length > 0) {
|
|
this.logger.log(`Marked pending messages as delivered for ${userId} across ${results.length} conversation(s)`);
|
|
}
|
|
})
|
|
.catch((err) => {
|
|
this.logger.warn(`Failed to mark pending messages as delivered for ${userId}: ${err.message}`);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Handle WebSocket disconnections
|
|
*/
|
|
async handleDisconnect(client: AuthenticatedSocket) {
|
|
if (client.userId) {
|
|
// Remove socket from local tracking
|
|
const sockets = this.userSocketMap.get(client.userId);
|
|
if (sockets) {
|
|
sockets.delete(client.id);
|
|
// Only mark offline if no more local sockets connected
|
|
if (sockets.size === 0) {
|
|
this.userSocketMap.delete(client.userId);
|
|
await this.redisPresence.setOffline(client.userId);
|
|
await this.messagesService.updateOnlineStatus(client.userId, false);
|
|
this.broadcastUserStatus(client.userId, false);
|
|
}
|
|
}
|
|
|
|
this.logger.log(`Client disconnected: ${client.id} (User: ${client.userId})`);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Join a conversation room
|
|
*/
|
|
@SubscribeMessage('join_conversation')
|
|
async handleJoinConversation(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { conversationId: string },
|
|
) {
|
|
if (!client.userId) {
|
|
return { error: 'Unauthorized' };
|
|
}
|
|
|
|
try {
|
|
// Verify user is part of the conversation
|
|
await this.messagesService.getConversationById(data.conversationId, client.userId);
|
|
|
|
// Join the room
|
|
client.join(`conversation:${data.conversationId}`);
|
|
|
|
this.logger.log(`User ${client.userId} joined conversation ${data.conversationId}`);
|
|
|
|
return { success: true };
|
|
} catch (error) {
|
|
this.logger.warn(`Join conversation failed for ${client.userId}: ${error.message}`);
|
|
return { error: 'Failed to join conversation' };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Leave a conversation room
|
|
*/
|
|
@SubscribeMessage('leave_conversation')
|
|
handleLeaveConversation(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { conversationId: string },
|
|
) {
|
|
client.leave(`conversation:${data.conversationId}`);
|
|
this.logger.log(`User ${client.userId} left conversation ${data.conversationId}`);
|
|
return { success: true };
|
|
}
|
|
|
|
/**
|
|
* Send a message via WebSocket
|
|
*/
|
|
@SubscribeMessage('send_message')
|
|
async handleSendMessage(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { conversationId: string; message: CreateMessageDto },
|
|
) {
|
|
if (!client.userId) {
|
|
return { error: 'Unauthorized' };
|
|
}
|
|
|
|
try {
|
|
let message = await this.messagesService.createMessage(
|
|
data.conversationId,
|
|
client.userId,
|
|
data.message,
|
|
);
|
|
|
|
// Send directly to the other participant's socket(s)
|
|
// This is the primary delivery mechanism — works even if they haven't joined the room.
|
|
// NOTE: We intentionally do NOT auto-mark as DELIVERED here based on Redis
|
|
// presence, because presence can lag behind reality (user turns off network
|
|
// but socket hasn't timed out yet). The receiver's client explicitly acks
|
|
// delivery via the `message_received` event once it actually receives
|
|
// `new_message` — that's the only reliable signal that the receiver's
|
|
// device got the payload.
|
|
try {
|
|
const participants = await this.messagesService.getConversationParticipants(data.conversationId);
|
|
if (participants) {
|
|
const otherUserId = client.userId === participants.userId
|
|
? participants.agentUserId
|
|
: participants.userId;
|
|
this.sendToUser(otherUserId, 'new_message', message);
|
|
}
|
|
} catch (participantError) {
|
|
this.logger.warn(`Failed to get participants for direct delivery in ${data.conversationId}: ${participantError.message}`);
|
|
}
|
|
|
|
// Also broadcast to the conversation room as a fallback
|
|
// (client.to excludes sender; recipients dedup by message ID)
|
|
client
|
|
.to(`conversation:${data.conversationId}`)
|
|
.emit('new_message', message);
|
|
|
|
return { success: true, message };
|
|
} catch (error) {
|
|
this.logger.warn(`Send message failed for ${client.userId}: ${error.message}`);
|
|
return { error: 'Failed to send message' };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle typing indicator start
|
|
*/
|
|
@SubscribeMessage('typing_start')
|
|
handleTypingStart(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { conversationId: string },
|
|
) {
|
|
if (!client.userId || !data?.conversationId) return;
|
|
|
|
// Rate limit typing events (max 1 per 2 seconds)
|
|
if (!this.checkRateLimit(client.userId, 'typing')) return;
|
|
|
|
// Broadcast to others in the room
|
|
client.to(`conversation:${data.conversationId}`).emit('typing_start', {
|
|
userId: client.userId,
|
|
conversationId: data.conversationId,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Handle typing indicator stop
|
|
*/
|
|
@SubscribeMessage('typing_stop')
|
|
handleTypingStop(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { conversationId: string },
|
|
) {
|
|
if (!client.userId || !data?.conversationId) return;
|
|
|
|
client.to(`conversation:${data.conversationId}`).emit('typing_stop', {
|
|
userId: client.userId,
|
|
conversationId: data.conversationId,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Acknowledge message delivery from receiver client.
|
|
* Called by mobile/web when `new_message` is received via socket so the
|
|
* sender gets a real-time double-tick regardless of Redis presence state.
|
|
*/
|
|
@SubscribeMessage('message_received')
|
|
async handleMessageReceived(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { messageId: string; conversationId: string },
|
|
) {
|
|
if (!client.userId || !data?.messageId) return { error: 'Invalid request' };
|
|
|
|
try {
|
|
// Verify the receiver is actually a participant of this conversation
|
|
const message = await this.prisma.message.findUnique({
|
|
where: { id: data.messageId },
|
|
select: { id: true, senderId: true, conversationId: true, status: true },
|
|
});
|
|
|
|
if (!message) return { error: 'Message not found' };
|
|
// Don't mark sender's own message as delivered
|
|
if (message.senderId === client.userId) return { success: false };
|
|
// Already delivered or read — no-op
|
|
if (message.status !== 'SENT') return { success: true };
|
|
|
|
const updated = await this.messagesService.markMessageDelivered(message.id);
|
|
const deliveredEvent = {
|
|
messageId: message.id,
|
|
conversationId: message.conversationId,
|
|
deliveredAt: (updated.deliveredAt ?? new Date()).toISOString(),
|
|
};
|
|
this.sendToUser(message.senderId, 'message_delivered', deliveredEvent);
|
|
return { success: true };
|
|
} catch (error) {
|
|
this.logger.warn(`message_received ack failed for ${client.userId}: ${error.message}`);
|
|
return { error: 'Failed to ack delivery' };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Mark messages as read
|
|
*/
|
|
@SubscribeMessage('mark_read')
|
|
async handleMarkRead(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { conversationId: string },
|
|
) {
|
|
if (!client.userId) {
|
|
return { error: 'Unauthorized' };
|
|
}
|
|
|
|
try {
|
|
await this.messagesService.markMessagesAsRead(data.conversationId, client.userId);
|
|
|
|
const readEvent = {
|
|
conversationId: data.conversationId,
|
|
readBy: client.userId,
|
|
readAt: new Date().toISOString(),
|
|
};
|
|
|
|
// Notify via room (for clients viewing this conversation)
|
|
client.to(`conversation:${data.conversationId}`).emit('messages_read', readEvent);
|
|
|
|
// Also notify the other participant directly
|
|
const participants = await this.messagesService.getConversationParticipants(data.conversationId);
|
|
if (participants) {
|
|
const otherUserId = client.userId === participants.userId
|
|
? participants.agentUserId
|
|
: participants.userId;
|
|
this.sendToUser(otherUserId, 'messages_read', readEvent);
|
|
}
|
|
|
|
return { success: true };
|
|
} catch (error) {
|
|
this.logger.warn(`Mark read failed for ${client.userId}: ${error.message}`);
|
|
return { error: 'Failed to mark as read' };
|
|
}
|
|
}
|
|
|
|
// ==========================================
|
|
// SUPPORT CHAT EVENTS
|
|
// ==========================================
|
|
|
|
/**
|
|
* Join a support chat room
|
|
*/
|
|
@SubscribeMessage('support_join')
|
|
async handleSupportJoin(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { chatId: string },
|
|
) {
|
|
if (!client.userId) {
|
|
return { error: 'Unauthorized' };
|
|
}
|
|
|
|
try {
|
|
const isAdmin = client.userRole === 'ADMIN';
|
|
await this.supportChatService.getChat(data.chatId, client.userId, isAdmin);
|
|
client.join(`support:${data.chatId}`);
|
|
this.logger.log(`User ${client.userId} joined support chat ${data.chatId}`);
|
|
return { success: true };
|
|
} catch (error) {
|
|
this.logger.warn(`Support join failed for ${client.userId}: ${error.message}`);
|
|
return { error: 'Failed to join support chat' };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Leave a support chat room
|
|
*/
|
|
@SubscribeMessage('support_leave')
|
|
handleSupportLeave(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { chatId: string },
|
|
) {
|
|
client.leave(`support:${data.chatId}`);
|
|
this.logger.log(`User ${client.userId} left support chat ${data.chatId}`);
|
|
return { success: true };
|
|
}
|
|
|
|
/**
|
|
* Send a message in a support chat via WebSocket
|
|
*/
|
|
@SubscribeMessage('support_send_message')
|
|
async handleSupportSendMessage(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { chatId: string; content: string },
|
|
) {
|
|
if (!client.userId) {
|
|
return { error: 'Unauthorized' };
|
|
}
|
|
|
|
try {
|
|
const senderRole = client.userRole === 'ADMIN' ? 'ADMIN' : 'USER';
|
|
const message = await this.supportChatService.sendMessage(
|
|
data.chatId,
|
|
client.userId,
|
|
senderRole,
|
|
data.content,
|
|
);
|
|
|
|
// Broadcast to all clients in the support chat room
|
|
this.server.to(`support:${data.chatId}`).emit('support_new_message', message);
|
|
|
|
// Notify admins about new unread count (for sidebar badge)
|
|
if (senderRole !== 'ADMIN') {
|
|
const unreadTotal = await this.supportChatService.getAdminUnreadTotal();
|
|
this.server.to('admin_room').emit('support_unread_update', { unreadTotal });
|
|
}
|
|
|
|
return { success: true, message };
|
|
} catch (error) {
|
|
this.logger.warn(`Support send message failed for ${client.userId}: ${error.message}`);
|
|
return { error: 'Failed to send message' };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Support chat typing indicators
|
|
*/
|
|
@SubscribeMessage('support_typing_start')
|
|
handleSupportTypingStart(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { chatId: string },
|
|
) {
|
|
if (!client.userId) return;
|
|
client.to(`support:${data.chatId}`).emit('support_typing_start', {
|
|
userId: client.userId,
|
|
chatId: data.chatId,
|
|
});
|
|
}
|
|
|
|
@SubscribeMessage('support_typing_stop')
|
|
handleSupportTypingStop(
|
|
@ConnectedSocket() client: AuthenticatedSocket,
|
|
@MessageBody() data: { chatId: string },
|
|
) {
|
|
if (!client.userId) return;
|
|
client.to(`support:${data.chatId}`).emit('support_typing_stop', {
|
|
userId: client.userId,
|
|
chatId: data.chatId,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Broadcast user online/offline status respecting activity_status privacy setting
|
|
*/
|
|
private async broadcastUserStatus(userId: string, isOnline: boolean) {
|
|
const statusEvent = {
|
|
userId,
|
|
isOnline,
|
|
lastSeenAt: isOnline ? null : new Date(),
|
|
};
|
|
|
|
try {
|
|
// Get user's activity_status privacy setting
|
|
const user = await this.prisma.user.findUnique({
|
|
where: { id: userId },
|
|
select: { privacyPreferences: true },
|
|
});
|
|
|
|
const prefs = user?.privacyPreferences as any;
|
|
const activityStatus = prefs?.privacySettings?.activity_status || 'public';
|
|
|
|
if (activityStatus === 'private') {
|
|
// Don't broadcast status to anyone
|
|
return;
|
|
}
|
|
|
|
if (activityStatus === 'connections') {
|
|
// Only send status to connected users
|
|
const connectedIds = await this.connectionRequestsService.getConnectedUserIds(userId);
|
|
for (const connectedId of connectedIds) {
|
|
this.sendToUser(connectedId, 'user_status_change', statusEvent);
|
|
}
|
|
return;
|
|
}
|
|
|
|
// 'public' — broadcast to everyone
|
|
this.server.emit('user_status_change', statusEvent);
|
|
} catch (err) {
|
|
// Fallback to public broadcast if privacy check fails
|
|
this.logger.warn(`Privacy check failed for ${userId}, broadcasting publicly: ${err.message}`);
|
|
this.server.emit('user_status_change', statusEvent);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Send message to a specific user via per-user room (Redis-backed cross-instance)
|
|
*/
|
|
sendToUser(userId: string, event: string, data: any) {
|
|
this.server.to(`user:${userId}`).emit(event, data);
|
|
this.logger.debug(`Sent ${event} to user:${userId} room`);
|
|
}
|
|
|
|
/**
|
|
* Check if a user is online (Redis-backed)
|
|
*/
|
|
async isUserOnline(userId: string): Promise<boolean> {
|
|
return this.redisPresence.isOnline(userId);
|
|
}
|
|
}
|