feat: implement rate limiting for socket events, sanitize error responses, and include agent profile details in user service output
This commit is contained in:
@@ -24,7 +24,7 @@ interface AuthenticatedSocket extends Socket {
|
|||||||
|
|
||||||
@WebSocketGateway({
|
@WebSocketGateway({
|
||||||
cors: {
|
cors: {
|
||||||
origin: '*', // Configure in production
|
origin: '*',
|
||||||
credentials: true,
|
credentials: true,
|
||||||
},
|
},
|
||||||
transports: ['websocket', 'polling'],
|
transports: ['websocket', 'polling'],
|
||||||
@@ -35,6 +35,7 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
|
|
||||||
private readonly logger = new Logger(MessagesGateway.name);
|
private readonly logger = new Logger(MessagesGateway.name);
|
||||||
private userSocketMap = new Map<string, Set<string>>(); // userId -> Set of socketIds
|
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(
|
constructor(
|
||||||
private readonly messagesService: MessagesService,
|
private readonly messagesService: MessagesService,
|
||||||
@@ -43,7 +44,30 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
private readonly prisma: PrismaService,
|
private readonly prisma: PrismaService,
|
||||||
private readonly jwtService: JwtService,
|
private readonly jwtService: JwtService,
|
||||||
private readonly configService: ConfigService,
|
private readonly configService: ConfigService,
|
||||||
) {}
|
) {
|
||||||
|
// 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
|
* Handle new WebSocket connections
|
||||||
@@ -169,7 +193,8 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
|
|
||||||
return { success: true };
|
return { success: true };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return { error: error.message };
|
this.logger.warn(`Join conversation failed for ${client.userId}: ${error.message}`);
|
||||||
|
return { error: 'Failed to join conversation' };
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -239,7 +264,8 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
|
|
||||||
return { success: true, message };
|
return { success: true, message };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return { error: error.message };
|
this.logger.warn(`Send message failed for ${client.userId}: ${error.message}`);
|
||||||
|
return { error: 'Failed to send message' };
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -251,7 +277,10 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
@ConnectedSocket() client: AuthenticatedSocket,
|
@ConnectedSocket() client: AuthenticatedSocket,
|
||||||
@MessageBody() data: { conversationId: string },
|
@MessageBody() data: { conversationId: string },
|
||||||
) {
|
) {
|
||||||
if (!client.userId) return;
|
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
|
// Broadcast to others in the room
|
||||||
client.to(`conversation:${data.conversationId}`).emit('typing_start', {
|
client.to(`conversation:${data.conversationId}`).emit('typing_start', {
|
||||||
@@ -268,7 +297,7 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
@ConnectedSocket() client: AuthenticatedSocket,
|
@ConnectedSocket() client: AuthenticatedSocket,
|
||||||
@MessageBody() data: { conversationId: string },
|
@MessageBody() data: { conversationId: string },
|
||||||
) {
|
) {
|
||||||
if (!client.userId) return;
|
if (!client.userId || !data?.conversationId) return;
|
||||||
|
|
||||||
client.to(`conversation:${data.conversationId}`).emit('typing_stop', {
|
client.to(`conversation:${data.conversationId}`).emit('typing_stop', {
|
||||||
userId: client.userId,
|
userId: client.userId,
|
||||||
@@ -311,7 +340,8 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
|
|
||||||
return { success: true };
|
return { success: true };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return { error: error.message };
|
this.logger.warn(`Mark read failed for ${client.userId}: ${error.message}`);
|
||||||
|
return { error: 'Failed to mark as read' };
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -338,7 +368,8 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
this.logger.log(`User ${client.userId} joined support chat ${data.chatId}`);
|
this.logger.log(`User ${client.userId} joined support chat ${data.chatId}`);
|
||||||
return { success: true };
|
return { success: true };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return { error: error.message };
|
this.logger.warn(`Support join failed for ${client.userId}: ${error.message}`);
|
||||||
|
return { error: 'Failed to join support chat' };
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -387,7 +418,8 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
|
|
||||||
return { success: true, message };
|
return { success: true, message };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return { error: error.message };
|
this.logger.warn(`Support send message failed for ${client.userId}: ${error.message}`);
|
||||||
|
return { error: 'Failed to send message' };
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -176,6 +176,24 @@ export class UsersService {
|
|||||||
country: profile.country,
|
country: profile.country,
|
||||||
}
|
}
|
||||||
: null,
|
: null,
|
||||||
|
agentProfile: user.agentProfile
|
||||||
|
? {
|
||||||
|
id: user.agentProfile.id,
|
||||||
|
slug: user.agentProfile.slug,
|
||||||
|
headline: user.agentProfile.headline,
|
||||||
|
bio: user.agentProfile.bio,
|
||||||
|
companyName: user.agentProfile.companyName,
|
||||||
|
licenseNumber: user.agentProfile.licenseNumber,
|
||||||
|
yearsOfExperience: user.agentProfile.yearsOfExperience,
|
||||||
|
isProfileComplete: user.agentProfile.isProfileComplete,
|
||||||
|
profileCompleteness: user.agentProfile.profileCompleteness,
|
||||||
|
agentTypeId: user.agentProfile.agentTypeId,
|
||||||
|
agentType: user.agentProfile.agentType,
|
||||||
|
isVerified: user.agentProfile.isVerified,
|
||||||
|
verificationStatus: user.agentProfile.verificationStatus,
|
||||||
|
verificationNote: user.agentProfile.verificationNote,
|
||||||
|
}
|
||||||
|
: null,
|
||||||
};
|
};
|
||||||
}),
|
}),
|
||||||
total,
|
total,
|
||||||
|
|||||||
Reference in New Issue
Block a user