Skip to content

Commit 47186c7

Browse files
codesailor4codesailor4
authored
Feat/websocket (#999)
* implemnted the websocket * implemnted the websocket * implemnted the websocket * implemnted the fix bootstrap * implemnted the search functionality
1 parent a635f24 commit 47186c7

3 files changed

Lines changed: 182 additions & 24 deletions

File tree

‎src/notifications/notifications.gateway.ts‎

Lines changed: 124 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,9 @@ import {
1010
MessageBody,
1111
} from '@nestjs/websockets';
1212
import { Server, Socket } from 'socket.io';
13-
import { Logger } from '@nestjs/common';
13+
import { Logger, Inject, UnauthorizedException } from '@nestjs/common';
14+
import { AuthService } from '../auth/auth.service';
15+
import { AuthUserPayload } from '../auth/types/auth-user.type';
1416

1517
@WebSocketGateway({
1618
cors: {
@@ -25,18 +27,73 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco
2527
private logger: Logger = new Logger('NotificationsGateway');
2628
private userSockets = new Map<string, string[]>();
2729
private socketUsers = new Map<string, string>();
30+
private eventRateLimits = new Map<string, { count: number; resetTime: number }>();
31+
private readonly MAX_CONNECTIONS_PER_USER = 5;
32+
private readonly MAX_EVENTS_PER_MINUTE = 60;
33+
private readonly RATE_LIMIT_WINDOW = 60 * 1000; // 1 minute in milliseconds
2834

29-
handleConnection(client: Socket) {
30-
const userId = client.handshake.query.userId as string;
31-
if (userId) {
32-
const sockets = this.userSockets.get(userId) || [];
33-
sockets.push(client.id);
34-
this.userSockets.set(userId, sockets);
35+
constructor(@Inject(AuthService) private readonly authService: AuthService) {}
36+
37+
private extractTokenFromHandshake(client: Socket): string | null {
38+
// Try to get token from authorization header
39+
const authHeader = client.handshake.headers.authorization;
40+
if (authHeader) {
41+
const [scheme, token] = authHeader.split(' ');
42+
if (scheme === 'Bearer' && token) {
43+
return token;
44+
}
45+
}
46+
47+
// Fallback to query parameter for websocket connections that can't send headers
48+
const tokenFromQuery = client.handshake.query.token as string;
49+
if (tokenFromQuery) {
50+
return tokenFromQuery;
51+
}
52+
53+
return null;
54+
}
55+
56+
async handleConnection(client: Socket) {
57+
try {
58+
// Extract and validate JWT token
59+
const token = this.extractTokenFromHandshake(client);
60+
if (!token) {
61+
this.logger.warn(`Connection rejected: Missing authentication token (client: ${client.id})`);
62+
client.disconnect(true);
63+
return;
64+
}
65+
66+
let authUser: AuthUserPayload;
67+
try {
68+
authUser = await this.authService.validateAccessToken(token);
69+
} catch (error) {
70+
this.logger.warn(`Connection rejected: Invalid JWT token (client: ${client.id})`);
71+
client.disconnect(true);
72+
return;
73+
}
74+
75+
const userId = authUser.sub;
76+
77+
// Check max connections per user
78+
const currentConnections = this.userSockets.get(userId) || [];
79+
if (currentConnections.length >= this.MAX_CONNECTIONS_PER_USER) {
80+
this.logger.warn(`Connection rejected: User ${userId} exceeded max connections (${this.MAX_CONNECTIONS_PER_USER})`);
81+
client.disconnect(true);
82+
return;
83+
}
84+
85+
// Add the new connection
86+
currentConnections.push(client.id);
87+
this.userSockets.set(userId, currentConnections);
3588
this.socketUsers.set(client.id, userId);
3689

90+
// Join the user's private room
3791
client.join(`user:${userId}`);
3892

39-
this.logger.log(`User ${userId} connected (${client.id})`);
93+
this.logger.log(`User ${userId} connected (${client.id}). Active connections: ${currentConnections.length}/${this.MAX_CONNECTIONS_PER_USER}`);
94+
} catch (error) {
95+
this.logger.error(`Error during connection handling: ${error.message}`, error.stack);
96+
client.disconnect(true);
4097
}
4198
}
4299

@@ -52,12 +109,47 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco
52109
this.userSockets.delete(userId);
53110
}
54111
this.socketUsers.delete(client.id);
55-
this.logger.log(`User ${userId} disconnected (${client.id})`);
112+
// Clean up rate limit entry for this socket
113+
this.eventRateLimits.delete(client.id);
114+
this.logger.log(`User ${userId} disconnected (${client.id}). Remaining connections: ${sockets.length}`);
56115
}
57116
}
58117

118+
private checkRateLimit(clientId: string): boolean {
119+
const now = Date.now();
120+
const rateLimit = this.eventRateLimits.get(clientId);
121+
122+
if (!rateLimit) {
123+
// First event, initialize rate limit
124+
this.eventRateLimits.set(clientId, { count: 1, resetTime: now + this.RATE_LIMIT_WINDOW });
125+
return true;
126+
}
127+
128+
// Reset counter if window has expired
129+
if (now > rateLimit.resetTime) {
130+
rateLimit.count = 1;
131+
rateLimit.resetTime = now + this.RATE_LIMIT_WINDOW;
132+
return true;
133+
}
134+
135+
// Check if limit exceeded
136+
if (rateLimit.count >= this.MAX_EVENTS_PER_MINUTE) {
137+
this.logger.warn(`Rate limit exceeded for client ${clientId}`);
138+
return false;
139+
}
140+
141+
// Increment counter
142+
rateLimit.count++;
143+
return true;
144+
}
145+
59146
@SubscribeMessage('joinProperty')
60147
handleJoinProperty(@ConnectedSocket() client: Socket, @MessageBody() data: { propertyId: string }) {
148+
// Check rate limit
149+
if (!this.checkRateLimit(client.id)) {
150+
return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } };
151+
}
152+
61153
if (data?.propertyId) {
62154
client.join(`property:${data.propertyId}`);
63155
this.logger.log(`Client ${client.id} joined property room ${data.propertyId}`);
@@ -68,18 +160,29 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco
68160

69161
@SubscribeMessage('leaveProperty')
70162
handleLeaveProperty(@ConnectedSocket() client: Socket, @MessageBody() data: { propertyId: string }) {
163+
// Check rate limit
164+
if (!this.checkRateLimit(client.id)) {
165+
return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } };
166+
}
167+
71168
if (data?.propertyId) {
72169
client.leave(`property:${data.propertyId}`);
73170
this.logger.log(`Client ${client.id} left property room ${data.propertyId}`);
74171
return { event: 'leftProperty', data: { propertyId: data.propertyId } };
75172
}
173+
return { event: 'error', data: { message: 'propertyId is required' } };
76174
}
77175

78176
@SubscribeMessage('joinTransaction')
79177
handleJoinTransaction(
80178
@ConnectedSocket() client: Socket,
81179
@MessageBody() data: { transactionId: string },
82180
) {
181+
// Check rate limit
182+
if (!this.checkRateLimit(client.id)) {
183+
return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } };
184+
}
185+
83186
if (data?.transactionId) {
84187
client.join(`transaction:${data.transactionId}`);
85188
this.logger.log(`Client ${client.id} joined transaction room ${data.transactionId}`);
@@ -93,14 +196,25 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco
93196
@ConnectedSocket() client: Socket,
94197
@MessageBody() data: { transactionId: string },
95198
) {
199+
// Check rate limit
200+
if (!this.checkRateLimit(client.id)) {
201+
return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } };
202+
}
203+
96204
if (data?.transactionId) {
97205
client.leave(`transaction:${data.transactionId}`);
98206
return { event: 'leftTransaction', data: { transactionId: data.transactionId } };
99207
}
208+
return { event: 'error', data: { message: 'transactionId is required' } };
100209
}
101210

102211
@SubscribeMessage('joinUser')
103212
handleJoinUser(@ConnectedSocket() client: Socket, @MessageBody() data: { userId: string }) {
213+
// Check rate limit
214+
if (!this.checkRateLimit(client.id)) {
215+
return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } };
216+
}
217+
104218
const socketUserId = this.socketUsers.get(client.id);
105219
if (socketUserId && socketUserId === data?.userId) {
106220
client.join(`user:${data.userId}`);
@@ -175,4 +289,4 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco
175289
sendToAll(event: string, data: any) {
176290
this.server.emit(event, data);
177291
}
178-
}
292+
}

‎src/notifications/notifications.module.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import {
1515
import { PrismaModule } from '../database/prisma.module';
1616
import { EmailModule } from '../email/email.module';
1717
import { UsersModule } from '../users/users.module';
18+
import { AuthModule } from '../auth/auth.module';
1819

1920
@Module({
2021
imports: [PrismaModule, EmailModule, UsersModule, ConfigModule],
@@ -30,4 +31,4 @@ import { UsersModule } from '../users/users.module';
3031
],
3132
exports: [NotificationsService, SmsService, SmsProviderFactory],
3233
})
33-
export class NotificationsModule {}
34+
export class NotificationsModule {}

‎src/search/search.service.ts‎

Lines changed: 56 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -86,18 +86,61 @@ export class SearchService {
8686
const { page = 1, limit = 20 } = searchQuery.pagination || {};
8787
const { field = 'createdAt', order = 'desc' } = searchQuery.sort || {};
8888

89-
// Mock data for now - this would typically query the database
90-
const items: any[] = [];
91-
const total = 0;
92-
93-
// Generate facets
94-
const facets = await this.facetsService.buildFacets(items, [
95-
'propertyType',
96-
'status',
97-
'city',
98-
'state',
99-
'bedrooms',
100-
'bathrooms',
89+
// Calculate skip for pagination
90+
const skip = (page - 1) * limit;
91+
92+
// Execute both count and findMany queries in parallel for better performance
93+
const [total, items, facets] = await Promise.all([
94+
// Get total count of matching properties
95+
this.prisma.property.count({ where: whereClause }),
96+
97+
// Get paginated, sorted properties
98+
this.prisma.property.findMany({
99+
where: whereClause,
100+
orderBy: { [field]: order },
101+
skip,
102+
take: limit,
103+
select: {
104+
id: true,
105+
title: true,
106+
description: true,
107+
address: true,
108+
city: true,
109+
state: true,
110+
zipCode: true,
111+
price: true,
112+
bedrooms: true,
113+
bathrooms: true,
114+
squareFootage: true,
115+
propertyType: true,
116+
status: true,
117+
latitude: true,
118+
longitude: true,
119+
createdAt: true,
120+
updatedAt: true,
121+
images: true,
122+
},
123+
}),
124+
125+
// Generate facets from all matching properties (not just paginated)
126+
this.facetsService.buildFacets(await this.prisma.property.findMany({
127+
where: whereClause,
128+
select: {
129+
propertyType: true,
130+
status: true,
131+
city: true,
132+
state: true,
133+
bedrooms: true,
134+
bathrooms: true,
135+
},
136+
}), [
137+
'propertyType',
138+
'status',
139+
'city',
140+
'state',
141+
'bedrooms',
142+
'bathrooms',
143+
])
101144
]);
102145

103146
// Get suggestions
@@ -145,4 +188,4 @@ export class SearchService {
145188
async getPopularSearches(): Promise<string[]> {
146189
return this.analyticsService.getPopularSearches();
147190
}
148-
}
191+
}

0 commit comments

Comments
 (0)