add support for events

Author: stefangutica

src/crons/websocket/events.custom.gateway.ts

@@ -0,0 +1,101 @@
+import { WebSocketGateway, WebSocketServer, SubscribeMessage, ConnectedSocket, MessageBody } from '@nestjs/websockets';
+import { Server, Socket } from 'socket.io';
+import { UseFilters, UseGuards } from '@nestjs/common';
+import { OriginLogger } from '@multiversx/sdk-nestjs-common';
+
+import { QueryPagination } from 'src/common/entities/query.pagination';
+import { WsValidationPipe } from 'src/utils/ws-validation.pipe';
+import { WebsocketExceptionsFilter } from 'src/utils/ws-exceptions.filter';
+import { WsSubscriptionLimiterGuard } from 'src/utils/ws.subscription.limiter';
+import { RoomKeyGenerator } from './room.key.generator';
+import { EventsService } from 'src/endpoints/events/events.service';
+import { EventsCustomSubscribePayload } from 'src/endpoints/events/entities/events.custom.subscribe';
+import { EventsFilter } from 'src/endpoints/events/entities/events.filter';
+import { Events } from 'src/endpoints/events/entities/events';
+
+
+@UseFilters(WebsocketExceptionsFilter)
+@WebSocketGateway({ cors: { origin: '*' }, path: '/ws/subscription' })
+export class EventsCustomGateway {
+  private readonly logger = new OriginLogger(EventsCustomGateway.name);
+
+  // Prefix distinct pentru camerele de evenimente pentru a nu se suprapune cu txs
+  static keyPrefix = 'custom-events-';
+
+  @WebSocketServer()
+  server!: Server;
+
+  constructor(
+    private readonly eventsService: EventsService,
+  ) { }
+
+  @UseGuards(WsSubscriptionLimiterGuard)
+  @SubscribeMessage('subscribeCustomEvents')
+  async handleCustomSubscription(
+    @ConnectedSocket() client: Socket,
+    @MessageBody(new WsValidationPipe()) payload: EventsCustomSubscribePayload
+  ) {
+    const filterIdentifier = RoomKeyGenerator.deterministicStringify(payload);
+    const roomName = `${EventsCustomGateway.keyPrefix}${filterIdentifier}`;
+
+    if (!client.rooms.has(roomName)) {
+      await client.join(roomName);
+    }
+
+    return { status: 'success' };
+  }
+
+  @SubscribeMessage('unsubscribeCustomEvents')
+  async handleCustomUnsubscribe(
+    @ConnectedSocket() client: Socket,
+    @MessageBody(new WsValidationPipe()) payload: EventsCustomSubscribePayload
+  ) {
+    const filterIdentifier = RoomKeyGenerator.deterministicStringify(payload);
+    const roomName = `${EventsCustomGateway.keyPrefix}${filterIdentifier}`;
+
+    if (client.rooms.has(roomName)) {
+      await client.leave(roomName);
+    }
+
+    return { status: 'unsubscribed' };
+  }
+
+  async pushEventsForTimestampMs(timestampMs: number): Promise<void> {
+    try {
+      const timestamp = timestampMs / 1000;
+
+      const allEvents = await this.eventsService.getEvents(
+        new QueryPagination({ size: 10000 }),
+        new EventsFilter({ before: timestamp, after: timestamp }),
+      );
+
+      const eventsFilteredForBroadcast: Map<string, Events[]> = new Map();
+
+      for (const event of allEvents) {
+        const roomKeys = RoomKeyGenerator.generate(
+          EventsCustomGateway.keyPrefix,
+          event,
+          EventsCustomSubscribePayload,
+        );
+
+        for (const roomKey of roomKeys) {
+          if (this.server.sockets.adapter.rooms.has(roomKey)) {
+            if (!eventsFilteredForBroadcast.has(roomKey)) {
+              eventsFilteredForBroadcast.set(roomKey, []);
+            }
+            eventsFilteredForBroadcast.get(roomKey)!.push(event);
+          }
+        }
+      }
+
+      for (const [roomName] of eventsFilteredForBroadcast) {
+        this.server.to(roomName).emit("customEventUpdate", {
+          events: eventsFilteredForBroadcast.get(roomName),
+          timestampMs
+        });
+      }
+    } catch (error) {
+      this.logger.error(error);
+    }
+  }
+}

src/crons/websocket/room.key.generator.ts

@@ -1,3 +1,4 @@
+import { EventsCustomSubscribePayload } from 'src/endpoints/events/entities/events.custom.subscribe';
 import { TransactionCustomSubscribePayload } from 'src/endpoints/transactions/entities/dtos/transaction.custom.subscribe';
 
 export class RoomKeyGenerator {
@@ -61,6 +62,8 @@ export class RoomKeyGenerator {
     switch (targetClass) {
       case TransactionCustomSubscribePayload:
         return TransactionCustomSubscribePayload.getClassFields();
+      case EventsCustomSubscribePayload:
+        return EventsCustomSubscribePayload.getClassFields();
       default:
         console.warn(`RoomKeyGenerator: No manual key mapping found for class ${targetClass.name}`);
         return [];

src/crons/websocket/websocket.cron.service.ts

@@ -19,6 +19,7 @@ import { NetworkService } from 'src/endpoints/network/network.service';
 import { Stats } from 'src/endpoints/network/entities/stats';
 import { TransactionsCustomGateway } from './transaction.custom.gateway';
 import { ConnectionHandler } from './connection.handler';
+import { EventsCustomGateway } from './events.custom.gateway';
 
 @Injectable()
 @WebSocketGateway({ cors: { origin: '*' }, path: '/ws/subscription' })
@@ -38,6 +39,7 @@ export class WebsocketCronService {
     private readonly elasticService: ElasticService,
     private readonly networkService: NetworkService,
     private readonly transactionsCustomGateway: TransactionsCustomGateway,
+    private readonly eventsCustomGateway: EventsCustomGateway,
     private readonly connectionHandler: ConnectionHandler,
   ) { }
 
@@ -111,7 +113,10 @@ export class WebsocketCronService {
       await this.pollUntil(async () => await this.isElasticDataAvailableForTimestampMs(roundToProcessTimestampMs, stats), pollingDelay, pollingMaxAttempts);
 
       // call gateways to process logic for custom subscriptions
-      await this.transactionsCustomGateway.pushTransactionsForTimestampMs(roundToProcessTimestampMs);
+      await Promise.all([
+        this.transactionsCustomGateway.pushTransactionsForTimestampMs(roundToProcessTimestampMs),
+        this.eventsCustomGateway.pushEventsForTimestampMs(roundToProcessTimestampMs)
+      ]);
       roundToProcessTimestampMs += stats.refreshRate;
     }
     this.cacheService.setLocal(

src/crons/websocket/websocket.subscription.module.ts

@@ -14,6 +14,7 @@ import { EventsGateway } from './events.gateway';
 import { ConnectionHandler } from './connection.handler';
 import { RoundModule } from 'src/endpoints/rounds/round.module';
 import { TransactionsCustomGateway } from './transaction.custom.gateway';
+import { EventsCustomGateway } from './events.custom.gateway';
 
 @Module({
   imports: [
@@ -34,6 +35,7 @@ import { TransactionsCustomGateway } from './transaction.custom.gateway';
     PoolGateway,
     EventsGateway,
     TransactionsCustomGateway,
+    EventsCustomGateway,
   ],
 })
 export class WebsocketSubscriptionModule { }

src/endpoints/events/entities/events.custom.subscribe.ts

@@ -0,0 +1,19 @@
+import { IsOptional, IsString } from 'class-validator';
+
+export class EventsCustomSubscribePayload {
+  @IsOptional()
+  @IsString()
+  address?: string;
+
+  @IsOptional()
+  @IsString()
+  identifier?: string;
+
+  @IsOptional()
+  @IsString()
+  logAddress?: string;
+
+  public static getClassFields(): string[] {
+    return ['address', 'identifier', 'logAddress'];
+  }
+}