added metric per topics

Author: bogdan-rosianu

src/common/metrics/api.metrics.service.ts

@@ -23,6 +23,7 @@ export class ApiMetricsService {
   private static transactionsPendingResultsCounter: Counter<string>;
   private static batchUpdatesCounter: Counter<string>;
   private static subscriptionsConnectionsGauge: Gauge<string>;
+  private static subscriptionsTopicConnectionsGauge: Gauge<string>;
 
   constructor(
     private readonly apiConfigService: ApiConfigService,
@@ -40,6 +41,14 @@ export class ApiMetricsService {
       });
     }
 
+    if (!ApiMetricsService.subscriptionsTopicConnectionsGauge) {
+      ApiMetricsService.subscriptionsTopicConnectionsGauge = new Gauge({
+        name: 'websocket_subscriptions_topic_connections',
+        help: 'Unique websocket clients per topic',
+        labelNames: ['topic'],
+      });
+    }
+
     if (!ApiMetricsService.vmQueriesHistogram) {
       ApiMetricsService.vmQueriesHistogram = new Histogram({
         name: 'vm_query',
@@ -191,10 +200,16 @@ export class ApiMetricsService {
   }
 
   @OnEvent(MetricsEvents.SetWebsocketMetrics)
-  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number }) {
-    const { connectedClients } = payload;
+  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number; topics?: Record<string, number> }) {
+    const { connectedClients, topics } = payload;
 
     ApiMetricsService.subscriptionsConnectionsGauge.set(connectedClients);
+
+    if (topics) {
+      for (const [topic, count] of Object.entries(topics)) {
+        ApiMetricsService.subscriptionsTopicConnectionsGauge.set({ topic }, count);
+      }
+    }
   }
 
 

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

@@ -48,15 +48,52 @@ export class WebsocketCronService {
   @Cron('*/1 * * * * *')
   handleWebsocketMetrics() {
     const connectedClients = this.server.sockets.sockets.size ?? 0;
-    // TODO: add more metrics in the future
-    // const subscriptions: Record<string, number> = {};
 
-    // this.server.sockets.adapter.rooms.forEach((socketsSet, roomName) => {
-    //   subscriptions[roomName] = socketsSet.size;
-    // });
+    // Efficient unique-listener counts per topic
+    const adapter = this.server.sockets.adapter as any;
+    const sids: Map<string, Set<string>> = adapter?.sids ?? new Map();
+
+    const topicPrefixes: Array<{ key: string; prefix?: string; room?: string }> = [
+      { key: 'tx', prefix: TransactionsGateway.keyPrefix },
+      { key: 'customTx', prefix: TransactionsCustomGateway.keyPrefix },
+      { key: 'events', prefix: EventsGateway.keyPrefix },
+      { key: 'customEvents', prefix: EventsCustomGateway.keyPrefix },
+      { key: 'blocks', prefix: BlocksGateway.keyPrefix },
+      { key: 'pool', prefix: PoolGateway.keyPrefix },
+      { key: 'stats', room: 'statsRoom' },
+    ];
+
+    const topics: Record<string, number> = {};
+    for (const { key } of topicPrefixes) topics[key] = 0;
+
+    // Count unique sockets per prefix-based topic by scanning socket -> rooms map once
+    if (sids && sids.size > 0) {
+      for (const [, rooms] of sids) {
+        // Track whether this socket has been counted for a given topic key
+        const matched: Record<string, boolean> = {};
+
+        for (const roomName of rooms) {
+          for (const { key, prefix } of topicPrefixes) {
+            if (!prefix || matched[key]) continue;
+            if (roomName.startsWith(prefix)) {
+              topics[key] += 1;
+              matched[key] = true;
+            }
+          }
+        }
+      }
+    }
+
+    // Handle exact-room topics (like statsRoom) directly from rooms map
+    const rooms: Map<string, Set<string>> = adapter?.rooms ?? new Map();
+    const statsRoomSet = rooms.get('statsRoom');
+    if (statsRoomSet) {
+      topics['stats'] = statsRoomSet.size;
+    }
 
     this.eventEmitter.emit(MetricsEvents.SetWebsocketMetrics, {
       connectedClients,
+      topics,
     });
   }