feat(observability): instrument event and websocket handlers with OTEL

.changeset/admin-metrics-handler-instrumentation.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat: instrument event and websocket handlers with OpenTelemetry metrics

src/adapters/web-socket-adapter.ts

@@ -15,6 +15,7 @@ import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants
 import { attemptValidation } from '../utils/validation'
 import { ContextMetadataKey } from '../constants/base'
 import { createLogger } from '../factories/logger-factory'
+import { recordWebsocketConnectionClosed, recordWebsocketConnectionOpened } from '../telemetry/event-metrics'
 import { Event } from '../@types/event'
 import { getRemoteAddress } from '../utils/http'
 import { IRateLimiter } from '../@types/utils'
@@ -82,6 +83,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
       .on(WebSocketAdapterEvent.Message, this.sendMessage.bind(this))
 
     logger('client %s connected from %s', this.clientId, this.clientAddress.address)
+    recordWebsocketConnectionOpened()
 
     // NIP-42
     this.challenge = randomBytes(32).toString('base64url')
@@ -260,6 +262,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
   }
 
   private onClientClose() {
+    recordWebsocketConnectionClosed()
     this.alive = false
     this.subscriptions.clear()
     this.authenticatedPubkeys.clear()

src/handlers/event-message-handler.ts

@@ -31,7 +31,7 @@ import {
 import { IEventRepository, INip05VerificationRepository, IUserRepository } from '../@types/repositories'
 import { IEventStrategy, IMessageHandler } from '../@types/message-handlers'
 import { CacheAdmissionState } from '../constants/caching'
-import { createCommandResult } from '../utils/messages'
+import { createEventCommandResult } from '../telemetry/event-metrics'
 import { createLogger } from '../factories/logger-factory'
 import { Factory } from '../@types/base'
 import { ICacheAdapter } from '../@types/adapters'
@@ -63,13 +63,13 @@ export class EventMessageHandler implements IMessageHandler {
     let reason = await this.isEventValid(event)
     if (reason) {
       logger('event %s rejected: %s', event.id, reason)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, reason))
       return
     }
 
     if (isExpiredEvent(event)) {
       logger('event %s rejected: expired')
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'event is expired'))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, 'event is expired'))
       return
     }
 
@@ -79,43 +79,43 @@ export class EventMessageHandler implements IMessageHandler {
       logger('event %s rejected: rate-limited')
       this.webSocket.emit(
         WebSocketAdapterEvent.Message,
-        createCommandResult(event.id, false, 'rate-limited: slow down'),
+        createEventCommandResult(event.id, false, 'rate-limited: slow down'),
       )
       return
     }
 
     reason = this.canAcceptEvent(event)
     if (reason) {
       logger('event %s rejected: %s', event.id, reason)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, reason))
       return
     }
 
     reason = await this.isProtectedEventBlocked(event)
     if (reason) {
       logger('event %s rejected: %s', event.id, reason)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, reason))
       return
     }
 
     reason = await this.isBlockedByRequestToVanish(event)
     if (reason) {
       logger('event %s rejected: %s', event.id, reason)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, reason))
       return
     }
 
     reason = await this.isUserAdmitted(event)
     if (reason) {
       logger('event %s rejected: %s', event.id, reason)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, reason))
       return
     }
 
     reason = await this.checkNip05Verification(event)
     if (reason) {
       logger('event %s rejected: %s', event.id, reason)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, reason))
       return
     }
 
@@ -124,7 +124,7 @@ export class EventMessageHandler implements IMessageHandler {
     if (typeof strategy?.execute !== 'function') {
       this.webSocket.emit(
         WebSocketAdapterEvent.Message,
-        createCommandResult(event.id, false, 'error: event not supported'),
+        createEventCommandResult(event.id, false, 'error: event not supported'),
       )
       return
     }
@@ -134,7 +134,7 @@ export class EventMessageHandler implements IMessageHandler {
       this.processNip05Metadata(event)
     } catch (error) {
       logger.error('error handling message', message, error)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: unable to process event'))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, 'error: unable to process event'))
     }
   }
 
@@ -242,7 +242,9 @@ export class EventMessageHandler implements IMessageHandler {
     }
 
     const checkEmbedded = async (evt: Event, depth = 0): Promise<boolean> => {
-      if (depth > 10) return false // Prevent infinite loops or excessive recursion
+      if (depth > 10) {
+        return false // Prevent infinite loops or excessive recursion
+      }
       if ((evt.kind === EventKinds.REPOST || evt.kind === EventKinds.GENERIC_REPOST) && evt.content.length > 0) {
         try {
           const embedded = attemptValidation(eventSchema)(JSON.parse(evt.content)) as Event

src/handlers/event-strategies/default-event-strategy.ts

@@ -1,4 +1,4 @@
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { IEventRepository } from '../../@types/repositories'
@@ -17,7 +17,7 @@ export class DefaultEventStrategy implements IEventStrategy<Event, Promise<void>
   public async execute(event: Event): Promise<void> {
     logger('received event: %o', event)
     const count = await this.eventRepository.create(event)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
 
     if (count) {
       this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)

src/handlers/event-strategies/delete-event-strategy.ts

@@ -1,4 +1,4 @@
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { EventTags } from '../../constants/base'
@@ -31,7 +31,7 @@ export class DeleteEventStrategy implements IEventStrategy<Event, Promise<void>>
     }
 
     const count = await this.eventRepository.create(event)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
 
     if (count) {
       this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)

src/handlers/event-strategies/ephemeral-event-strategy.ts

@@ -1,4 +1,4 @@
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { IEventStrategy } from '../../@types/message-handlers'
@@ -12,7 +12,7 @@ export class EphemeralEventStrategy implements IEventStrategy<Event, Promise<voi
 
   public async execute(event: Event): Promise<void> {
     logger('received ephemeral event: %o', event)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, ''))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, ''))
     this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)
   }
 }

src/handlers/event-strategies/gift-wrap-event-strategy.ts

@@ -1,4 +1,4 @@
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { EventTags } from '../../constants/base'
@@ -21,12 +21,12 @@ export class GiftWrapEventStrategy implements IEventStrategy<Event, Promise<void
 
     const reason = this.validateGiftWrap(event)
     if (reason) {
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, `invalid: ${reason}`))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, `invalid: ${reason}`))
       return
     }
 
     const count = await this.eventRepository.create(event)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
 
     if (count) {
       this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)

src/handlers/event-strategies/group-event-strategy.ts

@@ -1,4 +1,4 @@
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { EventTags } from '../../constants/base'
@@ -20,12 +20,12 @@ export class GroupEventStrategy implements IEventStrategy<Event, Promise<void>>
 
     const reason = this.validateGroupEvent(event)
     if (reason) {
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, `invalid: ${reason}`))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, `invalid: ${reason}`))
       return
     }
 
     const count = await this.eventRepository.create(event)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
 
     if (count) {
       this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)

src/handlers/event-strategies/parameterized-replaceable-event-strategy.ts

@@ -1,6 +1,6 @@
 import { Event, ParameterizedReplaceableEvent } from '../../@types/event'
 import { EventDeduplicationMetadataKey, EventTags } from '../../constants/base'
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { IEventRepository } from '../../@types/repositories'
 import { IEventStrategy } from '../../@types/message-handlers'
@@ -29,7 +29,7 @@ export class ParameterizedReplaceableEventStrategy implements IEventStrategy<Eve
     }
 
     const count = await this.eventRepository.upsert(parameterizedReplaceableEvent)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
 
     if (count) {
       this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)

src/handlers/event-strategies/replaceable-event-strategy.ts

@@ -1,4 +1,4 @@
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { IEventRepository } from '../../@types/repositories'
@@ -18,7 +18,7 @@ export class ReplaceableEventStrategy implements IEventStrategy<Event, Promise<v
     logger('received replaceable event: %o', event)
     try {
       const count = await this.eventRepository.upsert(event)
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
       if (count) {
         this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)
       }
@@ -27,12 +27,12 @@ export class ReplaceableEventStrategy implements IEventStrategy<Event, Promise<v
         if (error.message.endsWith('duplicate key value violates unique constraint "events_event_id_unique"')) {
           this.webSocket.emit(
             WebSocketAdapterEvent.Message,
-            createCommandResult(event.id, false, 'rejected: event already exists'),
+            createEventCommandResult(event.id, false, 'rejected: event already exists'),
           )
           return
         }
 
-        this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: '))
+        this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, 'error: '))
       }
     }
   }

src/handlers/event-strategies/timestamp-event-strategy.ts

@@ -5,7 +5,7 @@ import { IEventRepository } from '../../@types/repositories'
 import { WebSocketAdapterEvent } from '../../constants/adapter'
 import { EventTags } from '../../constants/base'
 import { createLogger } from '../../factories/logger-factory'
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { validateOtsProof } from '../../utils/nip03'
 
 const debug = createLogger('timestamp-event-strategy')
@@ -36,12 +36,12 @@ export class TimestampEventStrategy implements IEventStrategy<Event, Promise<voi
 
     const reason = this.validate(event)
     if (reason) {
-      this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, `invalid: ${reason}`))
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, false, `invalid: ${reason}`))
       return
     }
 
     const count = await this.eventRepository.create(event)
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
 
     if (count) {
       this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)

src/handlers/event-strategies/vanish-event-strategy.ts

@@ -1,5 +1,5 @@
 import { IEventRepository, IUserRepository } from '../../@types/repositories'
-import { createCommandResult } from '../../utils/messages'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
 import { createLogger } from '../../factories/logger-factory'
 import { Event } from '../../@types/event'
 import { EventKinds } from '../../constants/base'
@@ -25,6 +25,6 @@ export class VanishEventStrategy implements IEventStrategy<Event, Promise<void>>
 
     await this.userRepository.setVanished(event.pubkey, true)
 
-    this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, count ? '' : 'duplicate:'))
+    this.webSocket.emit(WebSocketAdapterEvent.Message, createEventCommandResult(event.id, true, count ? '' : 'duplicate:'))
   }
 }

src/telemetry/event-metrics.ts

@@ -0,0 +1,23 @@
+import type { EventId } from '../@types/base'
+import { createCommandResult } from '../utils/messages'
+import { getRelayMetricInstruments } from './metrics'
+
+export const createEventCommandResult = (eventId: EventId, successful: boolean, message: string) => {
+  const instruments = getRelayMetricInstruments()
+
+  if (successful) {
+    instruments.eventsAcceptedTotal.add(1)
+  } else {
+    instruments.eventsRejectedTotal.add(1)
+  }
+
+  return createCommandResult(eventId, successful, message)
+}
+
+export const recordWebsocketConnectionOpened = (): void => {
+  getRelayMetricInstruments().websocketConnections.add(1)
+}
+
+export const recordWebsocketConnectionClosed = (): void => {
+  getRelayMetricInstruments().websocketConnections.add(-1)
+}