Websockets

config/config.devnet.yaml

@@ -5,7 +5,7 @@ api:
   publicPort: 3001
   private: true
   privatePort: 4001
-  websocket: true
+  websocket: false
 cron:
   cacheWarmer: true
   fastWarm: true

package.json

@@ -144,6 +144,8 @@
     "rxjs": "^7.1.0",
     "sharp": "^0.34.2",
     "simple-git": "^3.16.0",
+    "socket.io": "^4.8.1",
+    "socket.io-client": "^4.8.1",
     "swagger-ui-express": "^4.3.0",
     "tiny-async-pool": "^1.2.0",
     "typeorm": "^0.3.25",

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

@@ -0,0 +1,32 @@
+import { Injectable } from '@nestjs/common';
+import { Cron } from '@nestjs/schedule';
+import { TransactionsGateway } from '../../endpoints/transactions/transaction.gateway';
+import { BlocksGateway } from 'src/endpoints/blocks/blocks.gateway';
+import { NetworkGateway } from 'src/endpoints/network/network.gateway';
+import { Lock } from "@multiversx/sdk-nestjs-common";
+@Injectable()
+export class WebsocketCronService {
+  constructor(
+    private readonly transactionsGateway: TransactionsGateway,
+    private readonly blocksGateway: BlocksGateway,
+    private readonly networkGateway: NetworkGateway,
+  ) { }
+
+  @Cron('*/6 * * * * *')
+  @Lock({ name: 'Push transactions to subscribers', verbose: true })
+  async handleTransactionsUpdate() {
+    await this.transactionsGateway.pushTransactions();
+  }
+
+  @Cron('*/6 * * * * *')
+  @Lock({ name: 'Push blocks to subscribers', verbose: true })
+  async handleBlocksUpdate() {
+    await this.blocksGateway.pushBlocks();
+  }
+
+  @Cron('*/6 * * * * *')
+  @Lock({ name: 'Push stats to subscribers', verbose: true })
+  async handleStatsUpdate() {
+    await this.networkGateway.pushStats();
+  }
+}

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

@@ -0,0 +1,19 @@
+import { Module } from '@nestjs/common';
+import { ScheduleModule } from '@nestjs/schedule';
+import { TransactionModule } from 'src/endpoints/transactions/transaction.module';
+import { WebsocketCronService } from './websocket.cron.service';
+import { BlockModule } from 'src/endpoints/blocks/block.module';
+import { NetworkModule } from 'src/endpoints/network/network.module';
+
+@Module({
+  imports: [
+    ScheduleModule.forRoot(),
+    TransactionModule,
+    BlockModule,
+    NetworkModule,
+  ],
+  providers: [
+    WebsocketCronService,
+  ],
+})
+export class WebSocketCronModule { }

src/endpoints/blocks/block.module.ts

@@ -3,6 +3,7 @@ import { BlsModule } from "../bls/bls.module";
 import { IdentitiesModule } from "../identities/identities.module";
 import { NodeModule } from "../nodes/node.module";
 import { BlockService } from "./block.service";
+import { BlocksGateway } from "./blocks.gateway";
 
 @Module({
   imports: [
@@ -11,10 +12,10 @@ import { BlockService } from "./block.service";
     forwardRef(() => IdentitiesModule),
   ],
   providers: [
-    BlockService,
+    BlockService, BlocksGateway,
   ],
   exports: [
-    BlockService,
+    BlockService, BlocksGateway,
   ],
 })
 export class BlockModule { }

src/endpoints/blocks/blocks.gateway.ts

@@ -0,0 +1,48 @@
+import { WebSocketGateway, WebSocketServer, SubscribeMessage, OnGatewayDisconnect } from '@nestjs/websockets';
+import { Server, Socket } from 'socket.io';
+import { BlockService } from './block.service';
+import { BlockFilter } from './entities/block.filter';
+import { QueryPagination } from 'src/common/entities/query.pagination';
+
+@WebSocketGateway({ cors: { origin: '*' } })
+export class BlocksGateway implements OnGatewayDisconnect {
+  @WebSocketServer()
+  server!: Server;
+
+  constructor(private readonly blockService: BlockService) { }
+
+  @SubscribeMessage('subscribeBlocks')
+  async handleSubscription(client: Socket, payload: any) {
+    const filterHash = JSON.stringify(payload);
+    await client.join(`block-${filterHash}`);
+  }
+
+  async pushBlocks() {
+    for (const [roomName] of this.server.sockets.adapter.rooms) {
+      if (!roomName.startsWith("block-")) continue;
+
+      const filterHash = roomName.replace("block-", "");
+      const filter = JSON.parse(filterHash);
+
+      const blockFilter = new BlockFilter({
+        shard: filter.shard,
+        proposer: filter.proposer,
+        validator: filter.validator,
+        epoch: filter.epoch,
+        nonce: filter.nonce,
+        hashes: filter.hashes,
+        order: filter.order,
+      });
+
+      const blocks = await this.blockService.getBlocks(
+        blockFilter,
+        new QueryPagination({ from: filter.from || 0, size: filter.size || 25 }),
+        filter.withProposerIdentity,
+      );
+
+      this.server.to(roomName).emit('blocksUpdate', blocks);
+    }
+  }
+
+  handleDisconnect(_client: Socket) { }
+}

src/endpoints/network/network.gateway.ts

@@ -0,0 +1,25 @@
+import { WebSocketGateway, WebSocketServer, SubscribeMessage, OnGatewayDisconnect } from '@nestjs/websockets';
+import { Server, Socket } from 'socket.io';
+import { NetworkService } from './network.service';
+
+@WebSocketGateway({ cors: { origin: '*' } })
+export class NetworkGateway implements OnGatewayDisconnect {
+  @WebSocketServer()
+  server!: Server;
+
+  constructor(private readonly networkService: NetworkService) { }
+
+  @SubscribeMessage('subscribeStats')
+  async handleSubscription(client: Socket) {
+    await client.join('statsRoom');
+  }
+
+  async pushStats() {
+    if (this.server.sockets.adapter.rooms.has('statsRoom')) {
+      const stats = await this.networkService.getStats();
+      this.server.to('statsRoom').emit('statsUpdate', stats);
+    }
+  }
+
+  handleDisconnect(_client: Socket) { }
+}

src/endpoints/network/network.module.ts

@@ -8,6 +8,7 @@ import { TokenModule } from "../tokens/token.module";
 import { TransactionModule } from "../transactions/transaction.module";
 import { VmQueryModule } from "../vm.query/vm.query.module";
 import { NetworkService } from "./network.service";
+import { NetworkGateway } from "./network.gateway";
 
 @Module({
   imports: [
@@ -21,10 +22,10 @@ import { NetworkService } from "./network.service";
     forwardRef(() => SmartContractResultModule),
   ],
   providers: [
-    NetworkService,
+    NetworkService, NetworkGateway,
   ],
   exports: [
-    NetworkService,
+    NetworkService, NetworkGateway,
   ],
 })
 export class NetworkModule { }

src/endpoints/transactions/transaction.gateway.ts

@@ -0,0 +1,74 @@
+import { WebSocketGateway, WebSocketServer, SubscribeMessage, OnGatewayDisconnect } from '@nestjs/websockets';
+import { Server, Socket } from 'socket.io';
+import { TransactionService } from './transaction.service';
+import { TransactionFilter } from './entities/transaction.filter';
+import { QueryPagination } from 'src/common/entities/query.pagination';
+import { TransactionQueryOptions } from './entities/transactions.query.options';
+
+@WebSocketGateway({ cors: { origin: '*' } })
+export class TransactionsGateway implements OnGatewayDisconnect {
+  @WebSocketServer()
+  server!: Server;
+
+  constructor(private readonly transactionService: TransactionService) { }
+
+  @SubscribeMessage('subscribeTransactions')
+  async handleSubscription(client: Socket, payload: any) {
+    const filterHash = JSON.stringify(payload);
+    await client.join(`tx-${filterHash}`);
+  }
+
+  async pushTransactions() {
+    for (const [roomName] of this.server.sockets.adapter.rooms) {
+      if (!roomName.startsWith("tx-")) continue;
+
+      const filterHash = roomName.replace("tx-", "");
+      const filter = JSON.parse(filterHash);
+
+      const options = TransactionQueryOptions.applyDefaultOptions(filter.size || 25, {
+        withScResults: filter.withScResults,
+        withOperations: filter.withOperations,
+        withLogs: filter.withLogs,
+        withScamInfo: filter.withScamInfo,
+        withUsername: filter.withUsername,
+        withBlockInfo: filter.withBlockInfo,
+        withActionTransferValue: filter.withActionTransferValue,
+      });
+
+      const transactionFilter = new TransactionFilter({
+        sender: filter.sender,
+        receivers: filter.receiver,
+        token: filter.token,
+        functions: filter.functions,
+        senderShard: filter.senderShard,
+        receiverShard: filter.receiverShard,
+        miniBlockHash: filter.miniBlockHash,
+        hashes: filter.hashes,
+        status: filter.status,
+        before: filter.before,
+        after: filter.after,
+        condition: filter.condition,
+        order: filter.order,
+        relayer: filter.relayer,
+        isRelayed: filter.isRelayed,
+        isScCall: filter.isScCall,
+        round: filter.round,
+        withRelayedScresults: filter.withRelayedScresults,
+      });
+
+      TransactionFilter.validate(transactionFilter, filter.size || 25);
+
+      const txs = await this.transactionService.getTransactions(
+        transactionFilter,
+        new QueryPagination({ from: filter.from || 0, size: filter.size || 25 }),
+        options,
+        undefined,
+        filter.fields || [],
+      );
+
+      this.server.to(roomName).emit('transactionUpdate', txs);
+    }
+  }
+
+  handleDisconnect(_client: Socket) { }
+}