Implement WebSocket to Stream Silent Blocks

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",
     "bip32": "^2.0.0",

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/base-block-data-provider.abstract.ts

@@ -1,5 +1,6 @@
 import { OperationStateService } from '@/operation-state/operation-state.service';
 import { Logger } from '@nestjs/common';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 import {
     IndexerService,
     TransactionInput,
@@ -10,6 +11,7 @@ import { BlockStateService } from '@/block-state/block-state.service';
 import { BlockState } from '@/block-state/block-state.entity';
 
 export abstract class BaseBlockDataProvider<OperationState> {
+    protected readonly eventEmitter: EventEmitter2 = new EventEmitter2();
     protected abstract readonly logger: Logger;
     protected abstract readonly operationStateKey: string;
 

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

@@ -13,6 +13,7 @@ import {
 } from '@/block-data-providers/bitcoin-core/provider-fixtures';
 import { Test, TestingModule } from '@nestjs/testing';
 import { BlockStateService } from '@/block-state/block-state.service';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 
 describe('Bitcoin Core Provider', () => {
     let provider: BitcoinCoreProvider;
@@ -51,6 +52,10 @@ describe('Bitcoin Core Provider', () => {
                     provide: BlockStateService,
                     useClass: jest.fn(),
                 },
+                {
+                    provide: EventEmitter2,
+                    useValue: jest.fn(),
+                },
             ],
         }).compile();
 

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

@@ -29,6 +29,8 @@ import { AxiosRequestConfig } from 'axios';
 import * as currency from 'currency.js';
 import { AxiosRetryConfig, makeRequest } from '@/common/request';
 import { BlockStateService } from '@/block-state/block-state.service';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { INDEXED_BLOCK_EVENT } from '@/common/events';
 
 @Injectable()
 export class BitcoinCoreProvider
@@ -46,6 +48,7 @@ export class BitcoinCoreProvider
         indexerService: IndexerService,
         operationStateService: OperationStateService,
         blockStateService: BlockStateService,
+        protected readonly eventEmitter: EventEmitter2,
     ) {
         super(
             configService,
@@ -144,6 +147,8 @@ export class BitcoinCoreProvider
                     blockHash: blockHash,
                     blockHeight: height,
                 });
+
+                this.eventEmitter.emit(INDEXED_BLOCK_EVENT, height);
             }
         } finally {
             this.isSyncing = false;

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

@@ -9,13 +9,15 @@ import { BitcoinCoreProvider } from '@/block-data-providers/bitcoin-core/provide
 import { EsploraProvider } from '@/block-data-providers/esplora/provider';
 import { BlockStateService } from '@/block-state/block-state.service';
 import { BlockStateModule } from '@/block-state/block-state.module';
+import { EventEmitter2, EventEmitterModule } from '@nestjs/event-emitter';
 
 @Module({
     imports: [
         OperationStateModule,
         IndexerModule,
         ConfigModule,
         BlockStateModule,
+        EventEmitterModule,
     ],
     controllers: [],
     providers: [
@@ -26,12 +28,14 @@ import { BlockStateModule } from '@/block-state/block-state.module';
                 IndexerService,
                 OperationStateService,
                 BlockStateService,
+                EventEmitter2,
             ],
             useFactory: (
                 configService: ConfigService,
                 indexerService: IndexerService,
                 operationStateService: OperationStateService,
                 blockStateService: BlockStateService,
+                eventEmitter: EventEmitter2,
             ) => {
                 switch (configService.get<ProviderType>('providerType')) {
                     case ProviderType.ESPLORA:
@@ -40,13 +44,15 @@ import { BlockStateModule } from '@/block-state/block-state.module';
                             indexerService,
                             operationStateService,
                             blockStateService,
+                            eventEmitter,
                         );
                     case ProviderType.BITCOIN_CORE_RPC:
                         return new BitcoinCoreProvider(
                             configService,
                             indexerService,
                             operationStateService,
                             blockStateService,
+                            eventEmitter,
                         );
                     default:
                         throw Error('unrecognised provider type in config');

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

@@ -13,6 +13,8 @@ import {
 import { TAPROOT_ACTIVATION_HEIGHT } from '@/common/constants';
 import { BlockStateService } from '@/block-state/block-state.service';
 import { Cron, CronExpression } from '@nestjs/schedule';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { INDEXED_BLOCK_EVENT } from '@/common/events';
 
 @Injectable()
 export class EsploraProvider
@@ -31,6 +33,7 @@ export class EsploraProvider
         indexerService: IndexerService,
         operationStateService: OperationStateService,
         blockStateService: BlockStateService,
+        protected readonly eventEmitter: EventEmitter2,
     ) {
         super(
             configService,
@@ -175,6 +178,8 @@ export class EsploraProvider
                     blockHeight: height,
                     blockHash: hash,
                 });
+
+                this.eventEmitter.emit(INDEXED_BLOCK_EVENT, height);
             } catch (error) {
                 this.logger.error(
                     `Error processing transactions in block at height ${height}, hash ${hash}: ${error.message}`,

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

@@ -2,11 +2,13 @@ import { NestFactory } from '@nestjs/core';
 import { AppModule } from '@/app.module';
 import { ConfigService } from '@nestjs/config';
 import { LogLevel } from '@nestjs/common';
+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,36 @@
+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}`);
+    }
+
+    broadcastSilentBlock(silentBlock: Buffer) {
+        for (const client of this.server.clients) {
+            if (client.readyState === WebSocket.OPEN) {
+                client.send(silentBlock);
+            }
+        }
+    }
+}

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.spec.ts

@@ -2,6 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
 import { TransactionsService } from '@/transactions/transactions.service';
 import { SilentBlocksService } from '@/silent-blocks/silent-blocks.service';
 import { silentBlockEncodingFixture } from '@/silent-blocks/silent-blocks.service.fixtures';
+import { SilentBlocksGateway } from '@/silent-blocks/silent-blocks.gateway';
 import { DataSource, Repository } from 'typeorm';
 import { Transaction } from '@/transactions/transaction.entity';
 import { getRepositoryToken } from '@nestjs/typeorm';
@@ -31,6 +32,10 @@ describe('SilentBlocksService', () => {
                     provide: getRepositoryToken(Transaction),
                     useValue: transactionRepository,
                 },
+                {
+                    provide: SilentBlocksGateway,
+                    useValue: jest.fn(),
+                },
             ],
         }).compile();
 

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