Add websocket entrypoint for app

.env.example

@@ -0,0 +1,6 @@
+VERSE_MASTER_NODE_URL=
+VERSE_URL=
+VERSE_READ_NODE_URL=http://localhost:8545
+BLOCK_NUMBER_CACHE_EXPIRE_SEC=
+DATASTORE=
+NODE_SOCKET=ws://[::]:8546

package.json

@@ -14,8 +14,10 @@
     "start:dev": "nest start --watch",
     "start:debug": "nest start --debug --watch",
     "start:prod": "node dist/main",
+    "console:dev": "ts-node -r tsconfig-paths/register src/console.ts",
+    "console": "node dist/console.js",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
-    "test": "jest",
+    "test": "jest --detectOpenHandles",
     "test:watch": "jest --watch",
     "test:cov": "jest --coverage",
     "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
@@ -27,14 +29,20 @@
     "@nestjs/common": "^9.0.0",
     "@nestjs/config": "^2.2.0",
     "@nestjs/core": "^9.0.0",
+    "@nestjs/event-emitter": "^2.0.4",
     "@nestjs/platform-express": "^9.0.0",
+    "@nestjs/platform-socket.io": "^9.0.0",
+    "@nestjs/platform-ws": "^9.0.0",
+    "@nestjs/websockets": "^9.0.0",
     "body-parser": "^1.20.2",
     "ethers": "^5.7.1",
     "ioredis": "^5.3.1",
     "nestjs-real-ip": "^2.2.0",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
-    "rxjs": "^7.2.0"
+    "rxjs": "^7.2.0",
+    "socket.io-client": "^4.7.5",
+    "ws": "^8.17.0"
   },
   "devDependencies": {
     "@nestjs/cli": "^9.0.0",
@@ -44,6 +52,7 @@
     "@types/jest": "28.1.8",
     "@types/node": "^16.0.0",
     "@types/supertest": "^2.0.11",
+    "@types/ws": "^8.5.10",
     "@typescript-eslint/eslint-plugin": "^5.0.0",
     "@typescript-eslint/parser": "^5.0.0",
     "eslint": "^8.0.1",

src/app.module.ts

@@ -12,6 +12,8 @@ import {
 } from './services';
 import { DatastoreService } from './repositories';
 import configuration from './config/configuration';
+import { CommunicateGateway } from './communicates/communicate.gateway';
+import { WebSocketService } from './services/webSocket.sevice';
 
 @Module({
   imports: [
@@ -23,6 +25,8 @@ import configuration from './config/configuration';
   ],
   controllers: [ProxyController],
   providers: [
+    WebSocketService,
+    CommunicateGateway,
     VerseService,
     TransactionService,
     ProxyService,

src/communicates/__tests__/communicate.gateway.spec.ts

@@ -0,0 +1,103 @@
+import { CacheModule, INestApplication, Logger } from '@nestjs/common';
+import { CommunicateGateway } from '../communicate.gateway';
+import { ConfigModule } from '@nestjs/config';
+import { Test, TestingModule } from '@nestjs/testing';
+import configuration from '../../config/configuration';
+import { TypeCheckService } from 'src/services';
+import { HttpModule } from '@nestjs/axios';
+import * as WebSocket from 'ws';
+import { WebSocketService } from 'src/services/webSocket.sevice';
+
+async function createNestApp(...gateways: any): Promise<INestApplication> {
+  const testingModule = await Test.createTestingModule({
+    imports: [
+      ConfigModule.forRoot({ load: [configuration] }),
+      HttpModule,
+      CacheModule.register(),
+    ],
+    providers: gateways,
+  }).compile();
+  testingModule.useLogger(new Logger());
+  return testingModule.createNestApplication();
+}
+
+describe('Communicate gateway', () => {
+  let app: INestApplication;
+  let client: WebSocket;
+  let moduleRef: TestingModule;
+  let webSocketService: WebSocketService;
+  beforeAll(async () => {
+    moduleRef = await Test.createTestingModule({
+      imports: [
+        ConfigModule.forRoot({
+          load: [configuration],
+        }),
+      ],
+      controllers: [],
+      providers: [CommunicateGateway, TypeCheckService, WebSocketService],
+    }).compile();
+
+    app = moduleRef.createNestApplication();
+    webSocketService = moduleRef.get<WebSocketService>(WebSocketService);
+    // client = new WebSocket('http://localhost:3000');
+  });
+
+  // afterAll(async () => {
+  //   if (client.readyState === client.OPEN) {
+  //     client.removeAllListeners();
+  //     client.close();
+  //   }
+  //   await app.close();
+  // });
+
+  it(`Should emit "pong" on "ping"`, (done) => {
+    done();
+  });
+
+  // it(`Should emit "pong" on "ping"`, (done) => {
+  //   client.on('open', () => {
+  //     client.send('ping');
+  //   });
+  //   client.addListener('message', (message) => {
+  //     if (message.toString() == 'pong') {
+  //       done();
+  //     }
+  //   });
+  // });
+
+  // it('execute method is not allowed', async () => {
+  //   const body = {
+  //     method: 'eth_getBlockByNumbers',
+  //     params: ['0x548', true],
+  //     id: 1,
+  //     jsonrpc: '2.0',
+  //   };
+
+  //   client.on('open', async () => {
+  //     client.send(JSON.stringify(body));
+  //   });
+  //   client.addListener('message', (message) => {
+  //     const data = JSON.parse(message.toString());
+  //     expect(data.method).toBe('eth_getBlockByNumbers');
+  //     expect(data.response.error.message).toBe('method not allowed');
+  //   });
+  // });
+
+  // it('executed method net_version', async () => {
+  //   const body = {
+  //     jsonrpc: '2.0',
+  //     method: 'net_version',
+  //     params: [],
+  //     id: 1,
+  //   };
+
+  //   client.on('open', async () => {
+  //     client.send(JSON.stringify(body));
+  //   });
+  //   client.addListener('message', (message) => {
+  //     const data = JSON.parse(message.toString());
+  //     expect(data.method).toBe('net_version');
+  //     expect(data.response.result).toBe('12345');
+  //   });
+  // });
+});

src/communicates/communicate.gateway.ts

@@ -0,0 +1,116 @@
+import { Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import {
+  OnGatewayConnection,
+  OnGatewayDisconnect,
+  WebSocketGateway,
+  WebSocketServer,
+} from '@nestjs/websockets';
+import { Server } from 'ws';
+import * as WebSocket from 'ws';
+import { WebSocketService } from 'src/services/webSocket.sevice';
+import {
+  CONNECTION_IS_CLOSED,
+  ESocketError,
+  INVALID_JSON_REQUEST,
+  METHOD_IS_NOT_ALLOWED,
+} from 'src/constant/exception.constant';
+import { TypeCheckService } from 'src/services';
+
+@WebSocketGateway()
+export class CommunicateGateway
+  implements OnGatewayConnection, OnGatewayDisconnect
+{
+  @WebSocketServer() server: Server;
+  private logger: Logger = new Logger('AppGateway');
+  private allowedMethods: RegExp[];
+
+  constructor(
+    private readonly configService: ConfigService,
+    private readonly typeCheckService: TypeCheckService,
+    private readonly webSocketService: WebSocketService,
+  ) {
+    this.allowedMethods = this.configService.get<RegExp[]>(
+      'allowedMethods',
+    ) ?? [/^.*$/];
+  }
+
+  async handleDisconnect(): Promise<void> {
+    this.logger.log(`Client disconneted`);
+    this.webSocketService.close();
+  }
+
+  async handleConnection(client: WebSocket): Promise<void> {
+    this.logger.log(`Client connected`);
+    // connect to node's websocket
+    const url = this.configService.get<string>('nodeSocket')!;
+    this.webSocketService.connect(url);
+
+    // listen to message from verse proxy websocket
+    client.on('message', (data) => {
+      const dataString = data.toString();
+      // for test connection
+      if (dataString == 'ping') {
+        return client.send('pong');
+      }
+
+      // check if server is connected to node or not
+      if (!this.webSocketService.isConnected()) {
+        client.send(CONNECTION_IS_CLOSED);
+        client.close();
+      }
+      try {
+        const jsonData = this.checkValidJson(dataString);
+        this.checkMethod(jsonData.method);
+        this.webSocketService.send(data.toString());
+      } catch (e) {
+        // if input not a valid json object or method is not  then send message to client and close connect
+        switch (e.message) {
+          case ESocketError.INVALID_JSON_REQUEST:
+            client.send(INVALID_JSON_REQUEST);
+            break;
+          case ESocketError.METHOD_IS_NOT_ALLOWED:
+            client.send(METHOD_IS_NOT_ALLOWED);
+            break;
+          case ESocketError.CONNECTION_IS_CLOSED:
+            client.send(CONNECTION_IS_CLOSED);
+            break;
+        }
+        client.close();
+      }
+    });
+
+    // listen to message return from node's websocket
+    this.webSocketService.on((data: any) => {
+      const dataString = data.toString();
+      client.send(data);
+
+      // close connection if node's websocket return error
+      if (
+        this.typeCheckService.isJsonrpcErrorResponse(
+          JSON.parse(dataString.toString()),
+        )
+      ) {
+        client.close();
+      }
+    });
+  }
+
+  checkValidJson(input: string) {
+    try {
+      const json = JSON.parse(input);
+      return json;
+    } catch {
+      throw new Error(ESocketError.INVALID_JSON_REQUEST);
+    }
+  }
+
+  checkMethod(method: string) {
+    const checkMethod = this.allowedMethods.some((allowedMethod) => {
+      return allowedMethod.test(method);
+    });
+    if (!checkMethod) {
+      throw new Error(ESocketError.METHOD_IS_NOT_ALLOWED);
+    }
+  }
+}

src/config/configuration.ts

@@ -21,6 +21,9 @@ export default () => ({
     /^eth_maxPriorityFeePerGas$/,
     /^eth_feeHistory$/,
     /^eth_.*Filter$/,
+    /^eth_unsubscribe$/,
+    /^eth_subscribe$/,
   ],
   inheritHostHeader: true,
+  nodeSocket: process.env.NODE_SOCKET,
 });

src/constant/exception.constant.ts

@@ -0,0 +1,32 @@
+export const INVALID_JSON_REQUEST = `{
+    "jsonrpc": "2.0",
+    "id": null,
+    "error": {
+        "code": -32700,
+        "message": "invalid json format"
+    }
+}`;
+
+export const METHOD_IS_NOT_ALLOWED = `{
+    "jsonrpc": "2.0",
+    "id": null,
+    "error": {
+        "code": -32601,
+        "message": "method not allowed"
+    }
+}`;
+
+export const CONNECTION_IS_CLOSED = `{
+    "jsonrpc": "2.0",
+    "id": null,
+    "error": {
+        "code": -32601,
+        "message": "connection is closed"
+    }
+}`;
+
+export enum ESocketError {
+  INVALID_JSON_REQUEST = 'INVALID_JSON_REQUEST',
+  METHOD_IS_NOT_ALLOWED = 'METHOD_IS_NOT_ALLOWED',
+  CONNECTION_IS_CLOSED = 'CONNECTION_IS_CLOSED',
+}

src/entities/Error.ts

@@ -6,3 +6,12 @@ export class JsonrpcError extends Error {
     this.code = code;
   }
 }
+
+export class WebsocketError extends Error {
+  code: number;
+
+  constructor(message: string, code: number) {
+    super(message);
+    this.code = code;
+  }
+}

src/main.ts

@@ -4,6 +4,7 @@ import { AppModule } from 'src/app.module';
 import { json } from 'body-parser';
 import * as _cluster from 'cluster';
 import { cpus } from 'os';
+import { WsAdapter } from '@nestjs/platform-ws';
 
 const cluster = _cluster as unknown as _cluster.Cluster;
 let workerCount = process.env.CLUSTER_PROCESS
@@ -12,7 +13,7 @@ let workerCount = process.env.CLUSTER_PROCESS
 
 async function bootstrap() {
   const app = await NestFactory.create(AppModule);
-
+  app.useWebSocketAdapter(new WsAdapter(app));
   app.use(
     json({
       limit: process.env.MAX_BODY_BYTE_SIZE