refactor and notifier metrics #1

src/services/monitoring/matching.ts

@@ -18,7 +18,7 @@ export type NotificationReceiver = (message: XcmMessageNotify) => Promise<void>
 type SubLevel<TV> = AbstractSublevel<DB, Buffer | Uint8Array | string, string, TV>;
 
 export type ChainBlock = {
-  chainId: string | number,
+  chainId: string,
   blockHash: string,
   blockNumber: string
 }
@@ -64,6 +64,8 @@ export class MatchingEngine extends EventEmitter {
   async onOutboundMessage(outMsg: XcmMessageSent) {
     const log = this.#log;
 
+    this.emit(telemetry.Inbound, outMsg);
+
     // Confirmation key at destination
     await this.#mutex.runExclusive(async () => {
       const hashKey = `${outMsg.messageHash}:${outMsg.recipient}`;
@@ -91,7 +93,7 @@ export class MatchingEngine extends EventEmitter {
             .del(idKey)
             .del(hashKey)
             .write();
-          await this.#notify(outMsg, inMsg);
+          await this.#notifyMatched(outMsg, inMsg);
         } catch {
           log.info(
             '[%s:o] STORED hash=%s id=%s (subId=%s, block=%s #%s)',
@@ -119,7 +121,7 @@ export class MatchingEngine extends EventEmitter {
             outMsg.blockNumber
           );
           await this.#inbound.del(hashKey);
-          await this.#notify(outMsg, inMsg);
+          await this.#notifyMatched(outMsg, inMsg);
         } catch {
           log.info(
             '[%s:o] STORED hash=%s (subId=%s, block=%s #%s)',
@@ -138,6 +140,8 @@ export class MatchingEngine extends EventEmitter {
   async onInboundMessage(inMsg: XcmMessageReceived)  {
     const log = this.#log;
 
+    this.emit(telemetry.Inbound, inMsg);
+
     await this.#mutex.runExclusive(async () => {
       const hashKey = `${inMsg.messageHash}:${inMsg.chainId}`;
       const idKey = `${inMsg.messageId}:${inMsg.chainId}`;
@@ -154,7 +158,7 @@ export class MatchingEngine extends EventEmitter {
             inMsg.blockNumber
           );
           await this.#outbound.del(hashKey);
-          await this.#notify(outMsg, inMsg);
+          await this.#notifyMatched(outMsg, inMsg);
         } catch {
           log.info(
             '[%s:i] STORED hash=%s (subId=%s, block=%s #%s)',
@@ -189,7 +193,7 @@ export class MatchingEngine extends EventEmitter {
             .del(idKey)
             .del(hashKey)
             .write();
-          await this.#notify(outMsg, inMsg);
+          await this.#notifyMatched(outMsg, inMsg);
         } catch {
           log.info(
             '[%s:i] STORED hash=%s id=%s (subId=%s, block=%s #%s)',
@@ -227,19 +231,17 @@ export class MatchingEngine extends EventEmitter {
     await this.#mutex.waitForUnlock();
   }
 
-  async #notify(
+  async #notifyMatched(
     outMsg: XcmMessageSent,
     inMsg: XcmMessageReceived
   ) {
+    this.emit(telemetry.Matched, inMsg, outMsg);
+
     try {
       const message: XcmMessageNotify = new XcmMessageNotify(outMsg, inMsg);
       await this.#noticationReceiver(message);
-
-      this.emit(telemetry.Notify, message);
     } catch (e) {
       this.#log.error(e, 'Error on notification');
-
-      this.emit(telemetry.NotifyError, e);
     }
   }
 

src/services/monitoring/ops/dmp.ts

@@ -79,7 +79,7 @@ function createXcmMessageSent(
   {
     paraId, data, tx: {extrinsic}
   } : {
-  paraId: number,
+  paraId: string,
   data: Bytes,
   tx: types.TxWithIdAndEvent
 }) : GenericXcmMessageSentWithContext {
@@ -114,7 +114,7 @@ function findDmpMessages(api: ApiPromise) {
         if (paraIdStr) {
           return {
             tx,
-            paraId: parseInt(paraIdStr.replaceAll(',', '')),
+            paraId: paraIdStr.replaceAll(',', ''),
             beneficiary,
             assets
           };

src/services/monitoring/ops/ump.ts

@@ -100,7 +100,7 @@ function findOutboundUmpMessage(
                 return new GenericXcmMessageSentWithContext({
                   ...sentMsg,
                   messageData: data.toU8a(),
-                  recipient: 0, // always relay
+                  recipient: '0', // always relay
                   messageHash: xcmProgram.hash.toHex(),
                   messageId: getMessageId(xcmProgram),
                   instructions: xcmProgram.toHuman()

src/services/monitoring/ops/xcmp.ts

@@ -26,7 +26,7 @@ function findOutboundHrmpMessage(
   return (source: Observable<XcmMessageSentWithContext>)
   : Observable<GenericXcmMessageSentWithContext> => {
     return source.pipe(
-      mergeMap(sentMsg => {
+      mergeMap((sentMsg): Observable<GenericXcmMessageSentWithContext> => {
         const { blockHash, messageHash } = sentMsg;
         return getOutboundHrmpMessages(blockHash).pipe(
           map(messages =>  {
@@ -39,7 +39,7 @@ function findOutboundHrmpMessage(
                   new GenericXcmMessageSentWithContext({
                     ...sentMsg,
                     messageData: xcmProgram.toU8a(),
-                    recipient: recipient.toNumber(),
+                    recipient: recipient.toNumber().toString(),
                     messageHash: xcmProgram.hash.toHex(),
                     instructions: xcmProgram.toHuman(),
                     messageId: getMessageId(xcmProgram)

src/services/monitoring/switchboard.ts

@@ -194,6 +194,7 @@ export class Switchboard {
   collectTelemetry(collect: (observer: TelemetryObserver) => void) {
     collect({ id: TelemetrySources.engine, source: this.#engine});
     collect({ id: TelemetrySources.catcher, source: this.#catcher});
+    collect({ id: TelemetrySources.notifier, source: this.#notifier});
   }
 
   /**

src/services/monitoring/types.ts

@@ -44,7 +44,7 @@ export type XcmMessageWithContext = {
 
 export interface XcmMessageSentWithContext extends XcmMessageWithContext {
   messageData: Uint8Array,
-  recipient: number,
+  recipient: string,
   sender: AnyJson,
   instructions: AnyJson,
 }
@@ -121,7 +121,7 @@ export class XcmMessageReceived {
 
 export class GenericXcmMessageSentWithContext implements XcmMessageSentWithContext {
   messageData: Uint8Array;
-  recipient: number;
+  recipient: string;
   instructions: AnyJson;
   messageHash: HexString;
   event: AnyJson;
@@ -164,7 +164,7 @@ export class XcmMessageSent {
   subscriptionId: string;
   chainId: string;
   messageData: string;
-  recipient: number;
+  recipient: string;
   instructions: AnyJson;
   messageHash: HexString;
   event: AnyJson;

src/services/notification/hub.ts

@@ -1,4 +1,6 @@
-import { Logger, Services } from '../types.js';
+import EventEmitter from 'node:events';
+
+import { Logger, Services, TelementryNotifierEvents } from '../types.js';
 import { QuerySubscription, XcmMessageNotify } from '../monitoring/types.js';
 import { LogNotifier } from './log.js';
 import { Notifier } from './types.js';
@@ -9,18 +11,29 @@ import { WebhookNotifier } from './webhook.js';
  *
  * Provides resolution of the supported notifiers.
  */
-export class NotifierHub implements Notifier {
+export class NotifierHub extends EventEmitter implements Notifier {
   #log: Logger;
   #notifiers: {
     [property: string]: Notifier
   };
 
   constructor(services: Services) {
+    super();
+
     this.#log = services.log;
     this.#notifiers = {
       log: new LogNotifier(services),
       webhook: new WebhookNotifier(services)
     };
+
+    // delegate telemetry events
+    for (const n of Object.values(this.#notifiers)) {
+      for (const t of Object.values(TelementryNotifierEvents)) {
+        n.on(t, (...args: any[]) => {
+          this.emit(t, ...args);
+        });
+      }
+    }
   }
 
   async notify(sub: QuerySubscription, msg: XcmMessageNotify) {

src/services/notification/log.ts

@@ -1,11 +1,15 @@
+import EventEmitter from 'node:events';
+
 import { Logger, Services } from '../../services/types.js';
 import { QuerySubscription, XcmMessageNotify } from '../monitoring/types.js';
 import { Notifier } from './types.js';
 
-export class LogNotifier implements Notifier {
+export class LogNotifier extends EventEmitter implements Notifier {
   #log: Logger;
 
   constructor({ log }: Services) {
+    super();
+
     this.#log = log;
   }
 

src/services/notification/types.ts

@@ -1,5 +1,7 @@
+import EventEmitter from 'node:events';
+
 import { QuerySubscription, XcmMessageNotify } from '../monitoring/types.js';
 
-export interface Notifier {
+export interface Notifier extends EventEmitter {
   notify(sub: QuerySubscription, msg: XcmMessageNotify) : Promise<void>
 }

src/services/notification/webhook.ts

@@ -1,14 +1,14 @@
-import Stream from 'node:stream';
+import { EventEmitter } from 'node:events';
 
 import got from 'got';
 import { ulid } from 'ulidx';
 
 import version from '../../version.js';
 import { QuerySubscription, WebhookNotification, XcmMessageNotify } from '../monitoring/types.js';
-import { Logger, Services } from 'services/types.js';
+import { Logger, Services, TelementryNotifierEvents as telemetry } from '../types.js';
 
 import { Notifier } from './types.js';
-import { Scheduled, Scheduler } from 'services/persistence/scheduler.js';
+import { Scheduled, Scheduler } from '../persistence/scheduler.js';
 
 const DEFAULT_DELAY = 300000; // 5 minutes
 
@@ -22,7 +22,7 @@ const WebhookTaskType = 'task:webhook';
 
 export const Delivered = Symbol('delivered');
 
-export class WebhookNotifier extends Stream.EventEmitter implements Notifier {
+export class WebhookNotifier extends EventEmitter implements Notifier {
   #log: Logger;
   #scheduler: Scheduler;
 
@@ -106,17 +106,27 @@ export class WebhookNotifier extends Stream.EventEmitter implements Notifier {
           msg.origin.blockNumber,
           msg.destination.blockNumber
         );
+
         this.emit(Delivered, {
           id,
           msg
         });
+
+        this.emit(telemetry.Notify, {
+          type: config.type,
+          subscription: msg.subscriptionId,
+          origin: msg.origin.chainId,
+          destination: msg.destination.chainId,
+          outcome: msg.outcome,
+          sink: config.url
+        });
       } else {
         // Should not enter here, since the non success status codes
         // are retryable and will throw an exception when the limit
-        // is of retries is reached.
+        // of retries is reached.
         this.#log.error(
           'Not deliverable webhook %s %s',
-          config.url,
+          url,
           id
         );
       }
@@ -138,6 +148,16 @@ export class WebhookNotifier extends Stream.EventEmitter implements Notifier {
         'Scheduled webhook delivery %s',
         key
       );
+
+      this.emit(telemetry.NotifyError, {
+        type: config.type,
+        subscription: msg.subscriptionId,
+        origin: msg.origin.chainId,
+        destination: msg.destination.chainId,
+        outcome: msg.outcome,
+        sink: config.url,
+        error: 'max_retries'
+      });
     }
   }
 }

src/services/telemetry/exporters/catcher.ts

@@ -1,10 +1,10 @@
 import { Counter, Histogram } from 'prom-client';
 
 import {
-  TelementryCatcherEvents, TelemetryObserver
+  TelementryCatcherEvents as events, TelemetryObserver
 } from '../../types.js';
 
-export function catcherExports(
+export function catcherMetrics(
   { source }: TelemetryObserver
 ) {
   const timers : Record<string, () => void> = {};
@@ -36,8 +36,6 @@ export function catcherExports(
     labelNames: ['origin']
   });
 
-  const events = TelementryCatcherEvents;
-
   source.on(events.BlockCacheHit, ({ chainId }) => {
     blockCacheHitsCount.labels(chainId).inc();
   });