feat: ledger id null trigger

.env.demo

@@ -223,3 +223,5 @@ RESEND_API_KEY=re_xxxxxxxxxx
 
 # Prisma log type. Default set to error
 PRISMA_LOGS = error
+# DB_ALERT_ENABLE=
+# DB_ALERT_EMAILS=

.env.sample

@@ -242,3 +242,8 @@ RESEND_API_KEY=re_xxxxxxxxxx
 
 # Prisma log type. Ideally should have only error or warn enabled. Having query enabled can add a lot of unwanted logging for all types of queries being run
 # PRISMA_LOGS = error,warn,query
+
+# Comma separated emails that needs to be alerted in case the 'ledgerId' is set to null
+# DB_ALERT_EMAILS=
+# Boolean: to enable/disable db alerts. This needs the 'utility' microservice
+# DB_ALERT_ENABLE=

apps/api-gateway/src/main.ts

@@ -119,6 +119,14 @@ async function bootstrap(): Promise<void> {
   );
   app.useGlobalInterceptors(new NatsInterceptor());
   await app.listen(process.env.API_GATEWAY_PORT, `${process.env.API_GATEWAY_HOST}`);
-  Logger.log(`API Gateway is listening on port ${process.env.API_GATEWAY_PORT}`);
+  Logger.log(`API Gateway is listening on port ${process.env.API_GATEWAY_PORT}`, 'Success');
+
+  if ('true' === process.env.DB_ALERT_ENABLE?.trim()?.toLowerCase()) {
+    // in case it is enabled, log that
+    Logger.log(
+      "We have enabled DB alert for 'ledger_null' instances. This would send email in case the 'ledger_id' column in 'org_agents' table is set to null",
+      'DB alert enabled'
+    );
+  }
 }
 bootstrap();

apps/api-gateway/src/utilities/utilities.service.ts

@@ -3,14 +3,119 @@
 import { StoreObjectDto, UtilitiesDto } from './dtos/shortening-url.dto';
 import { NATSClient } from '@credebl/common/NATSClient';
 import { ClientProxy } from '@nestjs/microservices';
+import { Client as PgClient } from 'pg';
 
 @Injectable()
 export class UtilitiesService extends BaseService {
+  private readonly pg: PgClient;
+  private isSendingNatsAlert = false;
+
   constructor(
     @Inject('NATS_CLIENT') private readonly serviceProxy: ClientProxy,
     private readonly natsClient: NATSClient
   ) {
-    super('OrganizationService');
+    super('UtilitiesService');
+    if ('true' === process.env.DB_ALERT_ENABLE?.trim()?.toLowerCase()) {
+      if (!process.env.DATABASE_URL) {
+        throw new Error('DATABASE_URL environment variable is required');
+      } else {
+        this.pg = new PgClient({
+          connectionString: process.env.DATABASE_URL
+        });
+      }
+    }
+  }
+
+  async onModuleInit(): Promise<void> {
+    try {
+      if ('true' !== process.env.DB_ALERT_ENABLE?.trim()?.toLowerCase()) {
+        // in case it is not enabled, return
+        return;
+      }
+      await this.pg.connect();
+      await this.pg.query('LISTEN ledger_null');
+      this.logger.log('PostgreSQL notification listener connected');
+    } catch (err) {
+      this.logger.error(`Failed to connect PostgreSQL listener: ${err?.message}`);
+      throw err;
+    }
+
+    this.pg.on('notification', async (msg) => {
+      if ('true' !== process.env.DB_ALERT_ENABLE?.trim()?.toLowerCase()) {
+        // in case it is not enabled, return
+        return;
+      }
+
+      if ('ledger_null' === msg.channel) {
+        try {
+          if (this.isSendingNatsAlert) {
+            this.logger.warn('Skipping duplicate NATS alert send...');
+            return;
+          }
+
+          this.isSendingNatsAlert = true;
+
+          // Step 1: Count total records
+          const totalRes = await this.pg.query('SELECT COUNT(*) FROM org_agents');
+          const total = Number(totalRes.rows[0].count);
+
+          // If the org_agents table has no records, total will be 0, causing a division by zero resulting in Infinity or NaN
+          if (0 === total) {
+            this.logger.debug('No org_agents records found, skipping alert check');
+            return;
+          }
+
+          // Step 2: Count NULL ledgerId records
+          const nullRes = await this.pg.query('SELECT COUNT(*) FROM org_agents WHERE "ledgerId" IS NULL');
+          const nullCount = Number(nullRes.rows[0].count);
+
+          // Step 3: Calculate %
+          const percent = (nullCount / total) * 100;
+
+          // Condition: > 30%
+          if (30 >= percent) {
+            return;
+          }
+
+          const alertEmails =
+            process.env.DB_ALERT_EMAILS?.split(',')
+              .map((e) => e.trim())
+              .filter((e) => 0 < e.length) || [];
+
+          if (0 === alertEmails.length) {
+            this.logger.error(
+              `DB_ALERT_EMAILS is empty, skipping alert. There is a ${percent}% records are set to null for 'ledgerId' in 'org_agents' table`,
+              'DB alert'
+            );
+            return;
+          }
+
+          const emailDto = {
+            emailFrom: '',
+            emailTo: alertEmails,
+            emailSubject: '[ALERT] More than 30% org_agents ledgerId is NULL',
+            emailText: `ALERT: ${percent.toFixed(2)}% of org_agents records currently have ledgerId = NULL.`,
+            emailHtml: `<p><strong>ALERT:</strong> ${percent.toFixed(
+              2
+            )}% of <code>org_agents</code> have <code>ledgerId</code> = NULL.</p>`
+          };
+
+          const result = await this.natsClient.sendNatsMessage(this.serviceProxy, 'alert-db-ledgerId-null', {
+            emailDto
+          });
+          this.logger.debug('Received result', JSON.stringify(result, null, 2));
+        } catch (err) {
+          this.logger.error(err?.message ?? 'Error in ledgerId alert handler');
+        } finally {
+          // Once its done, reset the flag
+          this.isSendingNatsAlert = false;
+        }
+      }
+    });
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.pg?.end();
   }
 
   async createShorteningUrl(shorteningUrlDto: UtilitiesDto): Promise<string> {

apps/ledger/src/ledger.module.ts

@@ -2,7 +2,7 @@ import { Logger, Module } from '@nestjs/common';
 import { LedgerController } from './ledger.controller';
 import { LedgerService } from './ledger.service';
 import { SchemaModule } from './schema/schema.module';
-import { PrismaService } from '@credebl/prisma-service';
+import { PrismaServiceModule } from '@credebl/prisma-service';
 import { CredentialDefinitionModule } from './credential-definition/credential-definition.module';
 import { ClientsModule, Transport } from '@nestjs/microservices';
 import { LedgerRepository } from './repositories/ledger.repository';
@@ -16,23 +16,21 @@ import { ContextInterceptorModule } from '@credebl/context/contextInterceptorMod
 @Module({
   imports: [
     GlobalConfigModule,
-    LoggerModule, PlatformConfig, ContextInterceptorModule,
+    LoggerModule,
+    PlatformConfig,
+    ContextInterceptorModule,
     ClientsModule.register([
       {
         name: 'NATS_CLIENT',
         transport: Transport.NATS,
         options: getNatsOptions(CommonConstants.LEDGER_SERVICE, process.env.LEDGER_NKEY_SEED)
-
       }
     ]),
-    SchemaModule, CredentialDefinitionModule
+    SchemaModule,
+    CredentialDefinitionModule,
+    PrismaServiceModule
   ],
   controllers: [LedgerController],
-  providers: [
-    LedgerService,
-    PrismaService,
-    LedgerRepository,
-    Logger
-  ]
+  providers: [LedgerService, LedgerRepository, Logger]
 })
-export class LedgerModule { }
+export class LedgerModule {}

apps/utility/src/utilities.controller.ts

@@ -2,6 +2,7 @@ import { Controller, Logger } from '@nestjs/common';
 import { MessagePattern } from '@nestjs/microservices';
 import { UtilitiesService } from './utilities.service';
 import { IShorteningUrlData } from '../interfaces/shortening-url.interface';
+import { EmailDto } from '@credebl/common/dtos/email.dto';
 
 @Controller()
 export class UtilitiesController {
@@ -30,4 +31,17 @@ export class UtilitiesController {
       throw new Error('Error occured in Utility Microservices Controller');
     }
   }
+
+  @MessagePattern({ cmd: 'alert-db-ledgerId-null' })
+  async handleLedgerAlert(payload: { emailDto: EmailDto }): Promise<void> {
+    try {
+      this.logger.debug('Received msg in alert-db-service');
+      const result = await this.utilitiesService.handleLedgerAlert(payload.emailDto);
+      this.logger.debug('Received result in alert-db-service');
+      return result;
+    } catch (error) {
+      this.logger.error(error);
+      throw error;
+    }
+  }
 }

apps/utility/src/utilities.repository.ts

@@ -1,51 +1,60 @@
-import { PrismaService } from "@credebl/prisma-service";
-import { Injectable, Logger } from "@nestjs/common";
+import { PrismaService } from '@credebl/prisma-service';
+import { Injectable, Logger } from '@nestjs/common';
 // eslint-disable-next-line camelcase
-import { shortening_url } from "@prisma/client";
+import { platform_config, shortening_url } from '@prisma/client';
 
 @Injectable()
 export class UtilitiesRepository {
-    constructor(
-        private readonly prisma: PrismaService,
-        private readonly logger: Logger
-    ) { }
-
-    async saveShorteningUrl(
-        payload
-    ): Promise<object> {
-
-        try {
-
-            const { referenceId, invitationPayload } = payload;
-            const storeShorteningUrl = await this.prisma.shortening_url.upsert({
-                where: { referenceId },
-                update: { invitationPayload },
-                create: { referenceId, invitationPayload }
-            });
-
-            this.logger.log(`[saveShorteningUrl] - shortening url details ${referenceId}`);
-            return storeShorteningUrl;
-        } catch (error) {
-            this.logger.error(`Error in saveShorteningUrl: ${error} `);
-            throw error;
-        }
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: Logger
+  ) {}
+
+  async saveShorteningUrl(payload): Promise<object> {
+    try {
+      const { referenceId, invitationPayload } = payload;
+      const storeShorteningUrl = await this.prisma.shortening_url.upsert({
+        where: { referenceId },
+        update: { invitationPayload },
+        create: { referenceId, invitationPayload }
+      });
+
+      this.logger.log(`[saveShorteningUrl] - shortening url details ${referenceId}`);
+      return storeShorteningUrl;
+    } catch (error) {
+      this.logger.error(`Error in saveShorteningUrl: ${error} `);
+      throw error;
     }
-
-    // eslint-disable-next-line camelcase
-    async getShorteningUrl(referenceId): Promise<shortening_url> {
-        try {
-
-            const storeShorteningUrl = await this.prisma.shortening_url.findUnique({
-                where: {
-                    referenceId
-                }
-            });
-
-            this.logger.log(`[getShorteningUrl] - shortening url details ${referenceId}`);
-            return storeShorteningUrl;
-        } catch (error) {
-            this.logger.error(`Error in getShorteningUrl: ${error} `);
-            throw error;
+  }
+
+  // eslint-disable-next-line camelcase
+  async getShorteningUrl(referenceId): Promise<shortening_url> {
+    try {
+      const storeShorteningUrl = await this.prisma.shortening_url.findUnique({
+        where: {
+          referenceId
         }
+      });
+
+      this.logger.log(`[getShorteningUrl] - shortening url details ${referenceId}`);
+      return storeShorteningUrl;
+    } catch (error) {
+      this.logger.error(`Error in getShorteningUrl: ${error} `);
+      throw error;
+    }
+  }
+
+  /**
+   * Get platform config details
+   * @returns
+   */
+  // eslint-disable-next-line camelcase
+  async getPlatformConfigDetails(): Promise<platform_config> {
+    try {
+      return this.prisma.platform_config.findFirst();
+    } catch (error) {
+      this.logger.error(`[getPlatformConfigDetails] - error: ${JSON.stringify(error)}`);
+      throw error;
     }
-}
+  }
+}