Implement WebSocket to Stream Silent Blocks

config/dev.config.yaml

@@ -17,4 +17,3 @@ bitcoinCore:
  rpcPass: polarpass
  rpcUser: polaruser
  rpcPort: 18445
-

package.json

@@ -28,6 +28,7 @@
  "@nestjs/common": "^10.4.1",
  "@nestjs/config": "^3.2.3",
  "@nestjs/core": "^10.4.1",
+ "@nestjs/event-emitter": "^2.0.4",
  "@nestjs/microservices": "^10.4.1",
  "@nestjs/passport": "^10.0.0",
  "@nestjs/platform-express": "^10.3.7",
@@ -36,6 +37,7 @@
  "@nestjs/swagger": "^7.3.1",
  "@nestjs/typeorm": "^10.0.2",
  "@nestjs/websockets": "^10.3.7",
+ "@nestjs/platform-ws": "^10.4.4",
  "axios": "^1.7.2",
  "currency.js": "^2.0.4",
  "js-yaml": "^4.1.0",
@@ -55,6 +57,7 @@
  "@types/node": "18.15.11",
  "@types/secp256k1": "^4.0.6",
  "@types/supertest": "^2.0.11",
+ "@types/ws": "^8.5.12",
  "@typescript-eslint/eslint-plugin": "^5.0.0",
  "@typescript-eslint/parser": "^5.0.0",
  "class-transformer": "^0.5.1",

src/app.module.ts

@@ -9,10 +9,12 @@ import { SilentBlocksModule } from '@/silent-blocks/silent-blocks.module';
 import { OperationStateModule } from '@/operation-state/operation-state.module';
 import { ScheduleModule } from '@nestjs/schedule';
 import { BlockProviderModule } from '@/block-data-providers/block-provider.module';
+import { EventEmitterModule } from '@nestjs/event-emitter';
 
 @Module({
  imports: [
  ScheduleModule.forRoot(),
+ EventEmitterModule.forRoot(),
  ConfigModule.forRoot({
  ignoreEnvFile: true,
  load: [configuration],

src/block-data-providers/bitcoin-core/provider.ts

@@ -28,6 +28,8 @@ import {
 import { AxiosRequestConfig } from 'axios';
 import * as currency from 'currency.js';
 import { AxiosRetryConfig, makeRequest } from '@/common/request';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { INDEXED_BLOCK_EVENT } from '@/common/events';
 
 @Injectable()
 export class BitcoinCoreProvider
@@ -44,6 +46,7 @@ export class BitcoinCoreProvider
  private readonly configService: ConfigService,
  indexerService: IndexerService,
  operationStateService: OperationStateService,
+ private eventEmitter: EventEmitter2,
  ) {
  super(indexerService, operationStateService);
 
@@ -123,6 +126,8 @@ export class BitcoinCoreProvider
 
  state.indexedBlockHeight = height;
  await this.setState(state);
+
+ this.eventEmitter.emit(INDEXED_BLOCK_EVENT, height);
  }
  } finally {
  this.isSyncing = false;

src/block-data-providers/block-provider.module.ts

@@ -7,31 +7,45 @@ import { IndexerService } from '@/indexer/indexer.service';
 import { ProviderType } from '@/common/enum';
 import { BitcoinCoreProvider } from '@/block-data-providers/bitcoin-core/provider';
 import { EsploraProvider } from '@/block-data-providers/esplora/provider';
+import { EventEmitter2, EventEmitterModule } from '@nestjs/event-emitter';
 
 @Module({
- imports: [OperationStateModule, IndexerModule, ConfigModule],
+ imports: [
+ OperationStateModule,
+ IndexerModule,
+ ConfigModule,
+ EventEmitterModule,
+ ],
  controllers: [],
  providers: [
  {
  provide: 'BlockDataProvider',
- inject: [ConfigService, IndexerService, OperationStateService],
+ inject: [
+ ConfigService,
+ IndexerService,
+ OperationStateService,
+ EventEmitter2,
+ ],
  useFactory: (
  configService: ConfigService,
  indexerService: IndexerService,
  operationStateService: OperationStateService,
+ eventEmitter: EventEmitter2,
  ) => {
  switch (configService.get<ProviderType>('providerType')) {
  case ProviderType.ESPLORA:
  return new EsploraProvider(
  configService,
  indexerService,
  operationStateService,
+ eventEmitter,
  );
  case ProviderType.BITCOIN_CORE_RPC:
  return new BitcoinCoreProvider(
  configService,
  indexerService,
  operationStateService,
+ eventEmitter,
  );
  default:
  throw Error('unrecognised provider type in config');

src/block-data-providers/esplora/provider.ts

@@ -12,6 +12,7 @@
 } from '@/block-data-providers/esplora/interface';
 import { TAPROOT_ACTIVATION_HEIGHT } from '@/common/constants';
 import { Cron, CronExpression } from '@nestjs/schedule';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 
 @Injectable()
 export class EsploraProvider
@@ -29,6 +30,8 @@
  private readonly configService: ConfigService,
  indexerService: IndexerService,
  operationStateService: OperationStateService,
+ private eventEmitter: EventEmitter2,
+
  ) {
  super(indexerService, operationStateService);
 

src/common/common.ts

@@ -147,3 +147,7 @@ export const varIntSize = (value: number): number => {
  else if (value <= 0xffffffff) return 5;
  else return 9;
 };
+
+export const delay = (ms: number): Promise<void> => {
+ return new Promise((resolve) => setTimeout(resolve, ms));
+};

src/common/events.ts

@@ -0,0 +1 @@
+export const INDEXED_BLOCK_EVENT = 'INDEXED_BLOCK_EVENT';

src/main.ts

@@ -1,11 +1,13 @@
 import { NestFactory } from '@nestjs/core';
 import { AppModule } from '@/app.module';
 import { ConfigService } from '@nestjs/config';
+import { WsAdapter } from '@nestjs/platform-ws';
 
 declare const module: any;
 
 async function bootstrap() {
  const app = await NestFactory.create(AppModule);
+ app.useWebSocketAdapter(new WsAdapter(app));
 
  const configService = app.get<ConfigService>(ConfigService);
  const port = configService.get<number>('app.port');

src/silent-blocks/silent-blocks.gateway.ts

@@ -0,0 +1,37 @@
+import { Injectable, Logger } from '@nestjs/common';
+import {
+ WebSocketGateway,
+ WebSocketServer,
+ OnGatewayConnection,
+ OnGatewayDisconnect,
+} from '@nestjs/websockets';
+import { Server, WebSocket } from 'ws';
+
+@Injectable()
+@WebSocketGateway()
+export class SilentBlocksGateway
+ implements OnGatewayConnection, OnGatewayDisconnect
+{
+ private readonly logger = new Logger(SilentBlocksGateway.name);
+
+ @WebSocketServer() server: Server;
+
+ handleConnection(client: WebSocket) {
+ const remoteAddress = (client as any)._socket.remoteAddress;
+ this.logger.debug(`Client connected: ${remoteAddress}`);
+ }
+
+ handleDisconnect(client: WebSocket) {
+ const remoteAddress = (client as any)._socket.remoteAddress;
+ this.logger.debug(`Client disconnected: ${remoteAddress}`);
+ }
+
+ // Method to broadcast silent block to all connected clients
+ broadcastSilentBlock(silentBlock: Buffer) {
+ this.server.clients.forEach((client: WebSocket) => {
+ if (client.readyState === WebSocket.OPEN) {
+ client.send(silentBlock.toString('hex')); // Send silent block as hex string
+ }
+ });
+ }
+}

src/silent-blocks/silent-blocks.module.ts

@@ -4,10 +4,11 @@ import { Transaction } from '@/transactions/transaction.entity';
 import { TransactionsService } from '@/transactions/transactions.service';
 import { SilentBlocksController } from '@/silent-blocks/silent-blocks.controller';
 import { SilentBlocksService } from '@/silent-blocks/silent-blocks.service';
+import { SilentBlocksGateway } from '@/silent-blocks/silent-blocks.gateway';
 
 @Module({
  imports: [TypeOrmModule.forFeature([Transaction])],
- providers: [TransactionsService, SilentBlocksService],
+ providers: [TransactionsService, SilentBlocksService, SilentBlocksGateway],
  controllers: [SilentBlocksController],
  exports: [SilentBlocksService],
 })

src/silent-blocks/silent-blocks.service.ts

@@ -1,12 +1,28 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 import { Transaction } from '@/transactions/transaction.entity';
 import { TransactionsService } from '@/transactions/transactions.service';
 import { SILENT_PAYMENT_BLOCK_TYPE } from '@/common/constants';
-import { encodeVarInt, varIntSize } from '@/common/common';
+import { encodeVarInt, varIntSize, delay } from '@/common/common';
+import { SilentBlocksGateway } from '@/silent-blocks/silent-blocks.gateway';
+import { OnEvent } from '@nestjs/event-emitter';
+import { INDEXED_BLOCK_EVENT } from '@/common/events';
 
 @Injectable()
 export class SilentBlocksService {
- constructor(private readonly transactionsService: TransactionsService) {}
+ private readonly logger = new Logger(SilentBlocksService.name);
+
+ constructor(
+ private readonly transactionsService: TransactionsService,
+ private readonly silentBlocksGateway: SilentBlocksGateway,
+ ) {}
+
+ @OnEvent(INDEXED_BLOCK_EVENT)
+ async handleBlockIndexedEvent(blockHeight: number) {
+ this.logger.debug(`New block indexed: ${blockHeight}`);
+ await delay(1000);
+ const silentBlock = await this.getSilentBlockByHeight(blockHeight);
+ this.silentBlocksGateway.broadcastSilentBlock(silentBlock);
+ }
 
  private getSilentBlockLength(transactions: Transaction[]): number {
  let length = 1 + varIntSize(transactions.length); // 1 byte for type + varint for transactions count