Streaming metrics

src/controllers/api/streaming.ts

@@ -61,6 +61,8 @@ const LIMITER_LIMIT = 100;
 
 const limiter = new TTLCache<string, number>();
 
+const connections = new Set<WebSocket>();
+
 const streamingController: AppController = async (c) => {
   const upgrade = c.req.header('upgrade');
   const token = c.req.header('sec-websocket-protocol');
@@ -126,7 +128,8 @@ const streamingController: AppController = async (c) => {
   }
 
   socket.onopen = async () => {
-    streamingConnectionsGauge.inc();
+    connections.add(socket);
+    streamingConnectionsGauge.set(connections.size);
 
     if (!stream) return;
     const topicFilter = await topicToFilter(stream, c.req.query(), pubkey);
@@ -186,7 +189,8 @@ const streamingController: AppController = async (c) => {
   };
 
   socket.onclose = () => {
-    streamingConnectionsGauge.dec();
+    connections.delete(socket);
+    streamingConnectionsGauge.set(connections.size);
     controller.abort();
   };
 

src/metrics.ts

@@ -104,3 +104,8 @@ export const cachedLinkPreviewSizeGauge = new Gauge({
   name: 'ditto_cached_link_previews_size',
   help: 'Number of link previews in cache',
 });
+
+export const internalSubscriptionsSizeGauge = new Gauge({
+  name: 'ditto_internal_subscriptions_size',
+  help: "Number of active subscriptions to Ditto's internal relay",
+});

src/storages.ts

@@ -2,6 +2,7 @@
 import { Conf } from '@/config.ts';
 import { DittoDatabase } from '@/db/DittoDatabase.ts';
 import { DittoDB } from '@/db/DittoDB.ts';
+import { internalSubscriptionsSizeGauge } from '@/metrics.ts';
 import { AdminStore } from '@/storages/AdminStore.ts';
 import { EventsDB } from '@/storages/EventsDB.ts';
 import { SearchStore } from '@/storages/search-store.ts';
@@ -61,7 +62,7 @@ export class Storages {
   /** Internal pubsub relay between controllers and the pipeline. */
   public static async pubsub(): Promise<InternalRelay> {
     if (!this._pubsub) {
-      this._pubsub = Promise.resolve(new InternalRelay());
+      this._pubsub = Promise.resolve(new InternalRelay({ gauge: internalSubscriptionsSizeGauge }));
     }
     return this._pubsub;
   }

src/storages/InternalRelay.ts

@@ -10,10 +10,15 @@ import {
 } from '@nostrify/nostrify';
 import { Machina } from '@nostrify/nostrify/utils';
 import { matchFilter } from 'nostr-tools';
+import { Gauge } from 'prom-client';
 
 import { DittoEvent } from '@/interfaces/DittoEvent.ts';
 import { purifyEvent } from '@/utils/purify.ts';
 
+interface InternalRelayOpts {
+  gauge?: Gauge;
+}
+
 /**
  * PubSub event store for streaming events within the application.
  * The pipeline should push events to it, then anything in the application can subscribe to it.
@@ -21,6 +26,8 @@ import { purifyEvent } from '@/utils/purify.ts';
 export class InternalRelay implements NRelay {
   private subs = new Map<string, { filters: NostrFilter[]; machina: Machina<NostrEvent> }>();
 
+  constructor(private opts: InternalRelayOpts = {}) {}
+
   async *req(
     filters: NostrFilter[],
     opts?: { signal?: AbortSignal },
@@ -31,6 +38,7 @@ export class InternalRelay implements NRelay {
     yield ['EOSE', id];
 
     this.subs.set(id, { filters, machina });
+    this.opts.gauge?.set(this.subs.size);
 
     try {
       for await (const event of machina) {
@@ -38,6 +46,7 @@ export class InternalRelay implements NRelay {
       }
     } finally {
       this.subs.delete(id);
+      this.opts.gauge?.set(this.subs.size);
     }
   }