feat: implement silent block broadcasting

fixed event emitter issue

fix: Add a delay function for silent block events

fix: silent block broadcasting

Fix: Use a more explicit event name for indexing blocks

feat: implement silent block broadcasting

fix: Make event emitter protected

feat: implement silent block broadcasting

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",
     "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,12 +1,14 @@
 import { OperationStateService } from '@/operation-state/operation-state.service';
 import { Logger } from '@nestjs/common';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 import {
     IndexerService,
     TransactionInput,
     TransactionOutput,
 } from '@/indexer/indexer.service';
 
 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.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,
+        protected readonly 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,8 @@ import {
 } 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';
+import { INDEXED_BLOCK_EVENT } from '@/common/events';
 
 @Injectable()
 export class EsploraProvider
@@ -29,6 +31,7 @@ export class EsploraProvider
         private readonly configService: ConfigService,
         indexerService: IndexerService,
         operationStateService: OperationStateService,
+        protected readonly eventEmitter: EventEmitter2,
     ) {
         super(indexerService, operationStateService);
 
@@ -157,6 +160,8 @@ export class EsploraProvider
                 state.indexedBlockHeight = height;
                 state.lastProcessedTxIndex = i + this.batchSize - 1;
                 await this.setState(state);
+
+                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

@@ -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,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.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