diff --git a/src/notifications/notifications.gateway.ts b/src/notifications/notifications.gateway.ts index e02ab79a..fd28007d 100644 --- a/src/notifications/notifications.gateway.ts +++ b/src/notifications/notifications.gateway.ts @@ -10,7 +10,9 @@ import { MessageBody, } from '@nestjs/websockets'; import { Server, Socket } from 'socket.io'; -import { Logger } from '@nestjs/common'; +import { Logger, Inject, UnauthorizedException } from '@nestjs/common'; +import { AuthService } from '../auth/auth.service'; +import { AuthUserPayload } from '../auth/types/auth-user.type'; @WebSocketGateway({ cors: { @@ -25,18 +27,73 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco private logger: Logger = new Logger('NotificationsGateway'); private userSockets = new Map(); private socketUsers = new Map(); + private eventRateLimits = new Map(); + private readonly MAX_CONNECTIONS_PER_USER = 5; + private readonly MAX_EVENTS_PER_MINUTE = 60; + private readonly RATE_LIMIT_WINDOW = 60 * 1000; // 1 minute in milliseconds - handleConnection(client: Socket) { - const userId = client.handshake.query.userId as string; - if (userId) { - const sockets = this.userSockets.get(userId) || []; - sockets.push(client.id); - this.userSockets.set(userId, sockets); + constructor(@Inject(AuthService) private readonly authService: AuthService) {} + + private extractTokenFromHandshake(client: Socket): string | null { + // Try to get token from authorization header + const authHeader = client.handshake.headers.authorization; + if (authHeader) { + const [scheme, token] = authHeader.split(' '); + if (scheme === 'Bearer' && token) { + return token; + } + } + + // Fallback to query parameter for websocket connections that can't send headers + const tokenFromQuery = client.handshake.query.token as string; + if (tokenFromQuery) { + return tokenFromQuery; + } + + return null; + } + + async handleConnection(client: Socket) { + try { + // Extract and validate JWT token + const token = this.extractTokenFromHandshake(client); + if (!token) { + this.logger.warn(`Connection rejected: Missing authentication token (client: ${client.id})`); + client.disconnect(true); + return; + } + + let authUser: AuthUserPayload; + try { + authUser = await this.authService.validateAccessToken(token); + } catch (error) { + this.logger.warn(`Connection rejected: Invalid JWT token (client: ${client.id})`); + client.disconnect(true); + return; + } + + const userId = authUser.sub; + + // Check max connections per user + const currentConnections = this.userSockets.get(userId) || []; + if (currentConnections.length >= this.MAX_CONNECTIONS_PER_USER) { + this.logger.warn(`Connection rejected: User ${userId} exceeded max connections (${this.MAX_CONNECTIONS_PER_USER})`); + client.disconnect(true); + return; + } + + // Add the new connection + currentConnections.push(client.id); + this.userSockets.set(userId, currentConnections); this.socketUsers.set(client.id, userId); + // Join the user's private room client.join(`user:${userId}`); - this.logger.log(`User ${userId} connected (${client.id})`); + this.logger.log(`User ${userId} connected (${client.id}). Active connections: ${currentConnections.length}/${this.MAX_CONNECTIONS_PER_USER}`); + } catch (error) { + this.logger.error(`Error during connection handling: ${error.message}`, error.stack); + client.disconnect(true); } } @@ -52,12 +109,47 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco this.userSockets.delete(userId); } this.socketUsers.delete(client.id); - this.logger.log(`User ${userId} disconnected (${client.id})`); + // Clean up rate limit entry for this socket + this.eventRateLimits.delete(client.id); + this.logger.log(`User ${userId} disconnected (${client.id}). Remaining connections: ${sockets.length}`); } } + private checkRateLimit(clientId: string): boolean { + const now = Date.now(); + const rateLimit = this.eventRateLimits.get(clientId); + + if (!rateLimit) { + // First event, initialize rate limit + this.eventRateLimits.set(clientId, { count: 1, resetTime: now + this.RATE_LIMIT_WINDOW }); + return true; + } + + // Reset counter if window has expired + if (now > rateLimit.resetTime) { + rateLimit.count = 1; + rateLimit.resetTime = now + this.RATE_LIMIT_WINDOW; + return true; + } + + // Check if limit exceeded + if (rateLimit.count >= this.MAX_EVENTS_PER_MINUTE) { + this.logger.warn(`Rate limit exceeded for client ${clientId}`); + return false; + } + + // Increment counter + rateLimit.count++; + return true; + } + @SubscribeMessage('joinProperty') handleJoinProperty(@ConnectedSocket() client: Socket, @MessageBody() data: { propertyId: string }) { + // Check rate limit + if (!this.checkRateLimit(client.id)) { + return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } }; + } + if (data?.propertyId) { client.join(`property:${data.propertyId}`); this.logger.log(`Client ${client.id} joined property room ${data.propertyId}`); @@ -68,11 +160,17 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco @SubscribeMessage('leaveProperty') handleLeaveProperty(@ConnectedSocket() client: Socket, @MessageBody() data: { propertyId: string }) { + // Check rate limit + if (!this.checkRateLimit(client.id)) { + return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } }; + } + if (data?.propertyId) { client.leave(`property:${data.propertyId}`); this.logger.log(`Client ${client.id} left property room ${data.propertyId}`); return { event: 'leftProperty', data: { propertyId: data.propertyId } }; } + return { event: 'error', data: { message: 'propertyId is required' } }; } @SubscribeMessage('joinTransaction') @@ -80,6 +178,11 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco @ConnectedSocket() client: Socket, @MessageBody() data: { transactionId: string }, ) { + // Check rate limit + if (!this.checkRateLimit(client.id)) { + return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } }; + } + if (data?.transactionId) { client.join(`transaction:${data.transactionId}`); this.logger.log(`Client ${client.id} joined transaction room ${data.transactionId}`); @@ -93,14 +196,25 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco @ConnectedSocket() client: Socket, @MessageBody() data: { transactionId: string }, ) { + // Check rate limit + if (!this.checkRateLimit(client.id)) { + return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } }; + } + if (data?.transactionId) { client.leave(`transaction:${data.transactionId}`); return { event: 'leftTransaction', data: { transactionId: data.transactionId } }; } + return { event: 'error', data: { message: 'transactionId is required' } }; } @SubscribeMessage('joinUser') handleJoinUser(@ConnectedSocket() client: Socket, @MessageBody() data: { userId: string }) { + // Check rate limit + if (!this.checkRateLimit(client.id)) { + return { event: 'error', data: { message: 'Rate limit exceeded. Please try again later.' } }; + } + const socketUserId = this.socketUsers.get(client.id); if (socketUserId && socketUserId === data?.userId) { client.join(`user:${data.userId}`); @@ -175,4 +289,4 @@ export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisco sendToAll(event: string, data: any) { this.server.emit(event, data); } -} +} \ No newline at end of file diff --git a/src/notifications/notifications.module.ts b/src/notifications/notifications.module.ts index 9aac7755..59f5b94b 100644 --- a/src/notifications/notifications.module.ts +++ b/src/notifications/notifications.module.ts @@ -15,6 +15,7 @@ import { import { PrismaModule } from '../database/prisma.module'; import { EmailModule } from '../email/email.module'; import { UsersModule } from '../users/users.module'; +import { AuthModule } from '../auth/auth.module'; @Module({ imports: [PrismaModule, EmailModule, UsersModule, ConfigModule], @@ -30,4 +31,4 @@ import { UsersModule } from '../users/users.module'; ], exports: [NotificationsService, SmsService, SmsProviderFactory], }) -export class NotificationsModule {} +export class NotificationsModule {} \ No newline at end of file diff --git a/src/search/search.service.ts b/src/search/search.service.ts index 8511c0e2..f5dfdd0c 100644 --- a/src/search/search.service.ts +++ b/src/search/search.service.ts @@ -86,18 +86,61 @@ export class SearchService { const { page = 1, limit = 20 } = searchQuery.pagination || {}; const { field = 'createdAt', order = 'desc' } = searchQuery.sort || {}; - // Mock data for now - this would typically query the database - const items: any[] = []; - const total = 0; - - // Generate facets - const facets = await this.facetsService.buildFacets(items, [ - 'propertyType', - 'status', - 'city', - 'state', - 'bedrooms', - 'bathrooms', + // Calculate skip for pagination + const skip = (page - 1) * limit; + + // Execute both count and findMany queries in parallel for better performance + const [total, items, facets] = await Promise.all([ + // Get total count of matching properties + this.prisma.property.count({ where: whereClause }), + + // Get paginated, sorted properties + this.prisma.property.findMany({ + where: whereClause, + orderBy: { [field]: order }, + skip, + take: limit, + select: { + id: true, + title: true, + description: true, + address: true, + city: true, + state: true, + zipCode: true, + price: true, + bedrooms: true, + bathrooms: true, + squareFootage: true, + propertyType: true, + status: true, + latitude: true, + longitude: true, + createdAt: true, + updatedAt: true, + images: true, + }, + }), + + // Generate facets from all matching properties (not just paginated) + this.facetsService.buildFacets(await this.prisma.property.findMany({ + where: whereClause, + select: { + propertyType: true, + status: true, + city: true, + state: true, + bedrooms: true, + bathrooms: true, + }, + }), [ + 'propertyType', + 'status', + 'city', + 'state', + 'bedrooms', + 'bathrooms', + ]) ]); // Get suggestions @@ -145,4 +188,4 @@ export class SearchService { async getPopularSearches(): Promise { return this.analyticsService.getPopularSearches(); } -} +} \ No newline at end of file