fix: async flow and integrate prom-client

package.json

@@ -37,6 +37,7 @@
         "nats": "2.10.0",
         "notifme-sdk": "^1.16.13",
         "pg": "^8.2.1",
+        "prom-client": "^15.1.3",
         "reflect-metadata": "^0.1.13",
         "typeorm": "0.3.17",
         "winston": "^3.2.1"

src/common/metrics.ts

@@ -0,0 +1,28 @@
+import { Counter, Histogram, register } from "prom-client"
+
+export const successNotificationMetricsCounter = new Counter({
+    name: 'successful_notifications',
+    help: 'Number of successful notifications',
+})
+
+export const failedNotificationMetricsCounter = new Counter({
+    name: 'failed_notifications',
+    help: 'Number of failed notifications',
+})
+
+export const httpRequestMetricsCounter = new Counter({
+    name: 'http_requests_counter',
+    help: 'Number of requests on http endpoints',
+    labelNames: ['method', 'endpoint', 'statusCode']
+})
+
+export const natsHistogram = new Histogram({
+    name: 'nats_consumer_histogram',
+    help: 'nats consumer duration histogram',
+    labelNames: ['streamName', 'consumerName']
+})
+
+register.registerMetric(successNotificationMetricsCounter)
+register.registerMetric(failedNotificationMetricsCounter)
+register.registerMetric(httpRequestMetricsCounter)
+register.registerMetric(natsHistogram)

src/destination/destinationHandlers/sesHandler.ts

@@ -52,7 +52,7 @@ export class SESService implements Handler {
         this.mh = mh
     }
 
-    handle(event: Event, templates: NotificationTemplates[], setting: NotificationSettings, configsMap: Map<string, boolean>, destinationMap: Map<string, boolean>): boolean {
+    async handle(event: Event, templates: NotificationTemplates[], setting: NotificationSettings, configsMap: Map<string, boolean>, destinationMap: Map<string, boolean>): Promise<boolean> {
         let sesTemplate: NotificationTemplates = templates.find(t => {
             return 'ses' == t.channel_type
         })
@@ -65,7 +65,7 @@ export class SESService implements Handler {
         this.sesConfig = null
         for (const element of providersSet) {
             if (element['dest'] === "ses") {
-                this.getDefaultConfig(providersSet, event, sesTemplate, setting, destinationMap, configsMap)
+                await this.getDefaultConfig(providersSet, event, sesTemplate, setting, destinationMap, configsMap)
                 break
             }
         }
@@ -82,7 +82,7 @@ export class SESService implements Handler {
                 from_email: config['from_email']
             }
             if(this.sesConfig && this.sesConfig.from_email){
-                providersSet.forEach(p => {
+                for (const p of providersSet) {
                     if (p['dest'] == "ses") {
                         let userId = p['configId']
                         let recipient = p['recipient']
@@ -93,19 +93,19 @@ export class SESService implements Handler {
                             configKey = p['dest'] + '-' + userId
                         }
                         if (!configsMap.get(configKey)) {
-                            this.processNotification(userId, recipient, event, sesTemplate, setting, p, emailMap)
+                            await this.processNotification(userId, recipient, event, sesTemplate, setting, p, emailMap)
                             configsMap.set(configKey, true)
                         }
                     }
-                });
+                };
             }
         } catch (error) {
             this.logger.error('getDefaultConfig', error)
             throw new CustomError("Unable to send ses notification",500);
         }
     }
 
-    private preparePaylodAndSend(event: Event, sesTemplate: NotificationTemplates, setting: NotificationSettings, p: string){
+    private async preparePayloadAndSend(event: Event, sesTemplate: NotificationTemplates, setting: NotificationSettings, p: string){
         let sdk: NotifmeSdk = new NotifmeSdk({
             channels: {
                 email: {
@@ -125,49 +125,67 @@ export class SESService implements Handler {
         // let options = { allowUndefinedFacts: true }
         let conditions: string = p['rule']['conditions'];
         if (conditions) {
-            engine.addRule({conditions: conditions, event: event});
-            engine.run(event).then(e => {
-                this.sendNotification(event, sdk, sesTemplate.template_payload).then(result => {
-                    this.saveNotificationEventSuccessLog(result, event, p, setting);
-                }).catch((error) => {
-                    this.logger.error(error.message);
-                    this.saveNotificationEventFailureLog(event, p, setting);
-                });
-            })
+          engine.addRule({ conditions: conditions, event: event });
+          try {
+            await engine.run(event);
+            const result = await this.sendNotification(
+              event,
+              sdk,
+              sesTemplate.template_payload
+            );
+            await this.saveNotificationEventSuccessLog(
+              result,
+              event,
+              p,
+              setting
+            );
+          } catch (error: any) {
+            this.logger.error(error.message);
+            await this.saveNotificationEventFailureLog(event, p, setting);
+          }
         } else {
-            this.sendNotification(event, sdk, sesTemplate.template_payload).then(result => {
-                this.saveNotificationEventSuccessLog(result, event, p, setting);
-            }).catch((error) => {
-                this.logger.error(error.message);
-                this.saveNotificationEventFailureLog(event, p, setting);
-            });
+          try {
+            const result = await this.sendNotification(
+              event,
+              sdk,
+              sesTemplate.template_payload
+            );
+            await this.saveNotificationEventSuccessLog(
+              result,
+              event,
+              p,
+              setting
+            );
+          } catch (error: any) {
+            this.logger.error(error.message);
+            await this.saveNotificationEventFailureLog(event, p, setting);
+          }
         }
     }
 
-    private processNotification(userId: number, recipient: string, event: Event, sesTemplate: NotificationTemplates, setting: NotificationSettings, p: string, emailMap: Map<string, boolean>) {
+    private async processNotification(userId: number, recipient: string, event: Event, sesTemplate: NotificationTemplates, setting: NotificationSettings, p: string, emailMap: Map<string, boolean>) {
         if(userId) {
-            this.usersRepository.findByUserId(userId).then(user => {
-                if (!user) {
-                    this.logger.info('no user found for id - ' + userId)
-                    this.logger.info(event.correlationId)
-                    return
-                }
-                this.sendEmailIfNotDuplicate(user['email_id'], event, sesTemplate, setting, p, emailMap)
-            })
+            const user = await this.usersRepository.findByUserId(userId)
+            if (!user) {
+                this.logger.info('no user found for id - ' + userId)
+                this.logger.info(event.correlationId)
+                return
+            }
+            await this.sendEmailIfNotDuplicate(user['email_id'], event, sesTemplate, setting, p, emailMap)
         }else{
             if (!recipient) {
                 this.logger.error('recipient is blank')
                 return
             }
-            this.sendEmailIfNotDuplicate(recipient, event, sesTemplate, setting, p, emailMap)
+            await this.sendEmailIfNotDuplicate(recipient, event, sesTemplate, setting, p, emailMap)
         }
     }
 
-    private sendEmailIfNotDuplicate(recipient : string, event: Event, sesTemplate: NotificationTemplates, setting: NotificationSettings, p: string, emailMap: Map<string, boolean>) {
+    private async sendEmailIfNotDuplicate(recipient : string, event: Event, sesTemplate: NotificationTemplates, setting: NotificationSettings, p: string, emailMap: Map<string, boolean>) {
         if (!emailMap.get(recipient)) {
             emailMap.set(recipient, true)
             event.payload['toEmail'] = recipient
-            this.preparePaylodAndSend(event, sesTemplate, setting, p)
+            await this.preparePayloadAndSend(event, sesTemplate, setting, p)
         } else {
             this.logger.info('duplicate email filtered out')
         }
@@ -205,17 +223,17 @@ export class SESService implements Handler {
         }
     }
 
-    private saveNotificationEventSuccessLog(result: any, event: Event, p: any, setting: NotificationSettings) {
+    private async saveNotificationEventSuccessLog(result: any, event: Event, p: any, setting: NotificationSettings) {
         if (result["status"] == "error") {
-            this.saveNotificationEventFailureLog(event, p, setting)
+            await this.saveNotificationEventFailureLog(event, p, setting)
         } else {
             let eventLog = this.eventLogBuilder.buildEventLog(event, p.dest, true, setting);
-            this.eventLogRepository.saveEventLog(eventLog);
+            await this.eventLogRepository.saveEventLog(eventLog);
         }
     }
 
-    private saveNotificationEventFailureLog(event: Event, p: any, setting: NotificationSettings) {
+    private async saveNotificationEventFailureLog(event: Event, p: any, setting: NotificationSettings) {
         let eventLog = this.eventLogBuilder.buildEventLog(event, p.dest, false, setting);
-        this.eventLogRepository.saveEventLog(eventLog);
+        await this.eventLogRepository.saveEventLog(eventLog);
     }
 }

src/destination/destinationHandlers/slackHandler.ts

@@ -44,7 +44,7 @@ export class SlackService implements Handler {
         this.mh = mh;
     }
 
-    handle(event: Event, templates: NotificationTemplates[], setting: NotificationSettings, configsMap: Map<string, boolean>, destinationMap: Map<string, boolean>): boolean {
+    async handle(event: Event, templates: NotificationTemplates[], setting: NotificationSettings, configsMap: Map<string, boolean>, destinationMap: Map<string, boolean>): Promise<boolean> {
 
         let slackTemplate: NotificationTemplates = templates.find(t => {
             return 'slack' == t.channel_type
@@ -57,21 +57,21 @@ export class SlackService implements Handler {
         const providerObjects = setting.config
         const providersSet = new Set(providerObjects);
 
-        providersSet.forEach(p => {
+        for (const p of providersSet) {
             if (p['dest'] == "slack") {
                 let slackConfigId = p['configId']
                 let configKey = p['dest'] + '-' + slackConfigId
                 if (!configsMap.get(configKey)) {
-                    this.processNotification(slackConfigId, event, slackTemplate, setting, p, destinationMap)
+                    await this.processNotification(slackConfigId, event, slackTemplate, setting, p, destinationMap)
                     configsMap.set(configKey, true)
                 }
             }
-        });
+        }
         return true
     }
 
-    private processNotification(slackConfigId: number, event: Event, slackTemplate: NotificationTemplates, setting: NotificationSettings, p: string, webhookMap: Map<string, boolean>) {
-        this.slackConfigRepository.findBySlackConfigId(slackConfigId).then(config => {
+    private async processNotification(slackConfigId: number, event: Event, slackTemplate: NotificationTemplates, setting: NotificationSettings, p: string, webhookMap: Map<string, boolean>) {
+            const config = await this.slackConfigRepository.findBySlackConfigId(slackConfigId)
             if (!config) {
                 this.logger.info('no slack config found for event')
                 this.logger.info(event.correlationId)
@@ -97,30 +97,45 @@ export class SlackService implements Handler {
             });
             let engine = new Engine();
             // let options = { allowUndefinedFacts: true }
-            let conditions: string = p['rule']['conditions'];
+            let conditions: string = p["rule"]["conditions"];
             if (conditions) {
-                engine.addRule({conditions: conditions, event: event});
-                engine.run(event).then(e => {
-                    this.sendNotification(event, sdk, slackTemplate.template_payload).then(result => {
-                        this.saveNotificationEventSuccessLog(result, event, p, setting);
-                    }).catch((error) => {
-                        this.logger.error(error.message);
-                        this.saveNotificationEventFailureLog(event, p, setting);
-                    });
-                })
+              engine.addRule({ conditions: conditions, event: event });
+              try {
+                await engine.run(event);
+                const result = await this.sendNotification(
+                  event,
+                  sdk,
+                  slackTemplate.template_payload
+                );
+                await this.saveNotificationEventSuccessLog(
+                  result,
+                  event,
+                  p,
+                  setting
+                );
+              } catch (error: any) {
+                this.logger.error(error.message);
+                await this.saveNotificationEventFailureLog(event, p, setting);
+              }
             } else {
-                this.sendAndLogNotification(event, sdk, setting, p, slackTemplate);
+              await this.sendAndLogNotification(
+                event,
+                sdk,
+                setting,
+                p,
+                slackTemplate
+              );
             }
-        })
     }
 
-    public sendAndLogNotification(event: Event, sdk: NotifmeSdk, setting: NotificationSettings, p: any, slackTemplate: NotificationTemplates){
-        this.sendNotification(event, sdk, slackTemplate.template_payload).then(result => {
-            this.saveNotificationEventSuccessLog(result, event, p, setting);
-        }).catch((error) => {
+    public async sendAndLogNotification(event: Event, sdk: NotifmeSdk, setting: NotificationSettings, p: any, slackTemplate: NotificationTemplates){
+        try {
+            const result = await this.sendNotification(event, sdk, slackTemplate.template_payload)
+            await this.saveNotificationEventSuccessLog(result, event, p, setting);
+        } catch(error: any) {
             this.logger.error(error.message);
             this.saveNotificationEventFailureLog(event, p, setting);
-        });
+        };
     }
 
     public async sendNotification(event: Event, sdk: NotifmeSdk, template: string) {
@@ -149,17 +164,17 @@ export class SlackService implements Handler {
         }
     }
 
-    private saveNotificationEventSuccessLog(result: any, event: Event, p: any, setting: NotificationSettings) {
+    private async saveNotificationEventSuccessLog(result: any, event: Event, p: any, setting: NotificationSettings) {
         if (result["status"] == "error") {
-            this.saveNotificationEventFailureLog(event, p, setting)
+            await this.saveNotificationEventFailureLog(event, p, setting)
         } else {
             let eventLog = this.eventLogBuilder.buildEventLog(event, p.dest, true, setting);
-            this.eventLogRepository.saveEventLog(eventLog);
+            await this.eventLogRepository.saveEventLog(eventLog);
         }
     }
 
-    private saveNotificationEventFailureLog(event: Event, p: any, setting: NotificationSettings) {
+    private async saveNotificationEventFailureLog(event: Event, p: any, setting: NotificationSettings) {
         let eventLog = this.eventLogBuilder.buildEventLog(event, p.dest, false, setting);
-        this.eventLogRepository.saveEventLog(eventLog);
+        await this.eventLogRepository.saveEventLog(eventLog);
     }
 }