Feat/websocket

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<string, string[]>();
   private socketUsers = new Map<string, string>();
+  private eventRateLimits = new Map<string, { count: number; resetTime: number }>();
+  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,18 +160,29 @@ 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')
   handleJoinTransaction(
     @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);
   }
-}
+}

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 {}

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<string[]> {
     return this.analyticsService.getPopularSearches();
   }
-}
+}