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()?.toLocaleLowerCase()) {
+    // 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,81 @@
 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 pg: PgClient;
+
   constructor(
     @Inject('NATS_CLIENT') private readonly serviceProxy: ClientProxy,
     private readonly natsClient: NATSClient
   ) {
-    super('OrganizationService');
+    super('UtilitiesService');
+    this.pg = new PgClient({
+      connectionString: process.env.DATABASE_URL
+    });
+  }
+
+  async onModuleInit(): Promise<void> {
+    await this.pg.connect();
+
+    // Listen to the notification channel
+    await this.pg.query('LISTEN ledger_null');
+
+    // NATS is not available → skip silently
+    this.pg.on('notification', async (msg) => {
+      if ('true' !== process.env.DB_ALERT_ENABLE?.trim()?.toLocaleLowerCase()) {
+        // in case it is not enabled, return
+        return;
+      }
+
+      if ('ledger_null' === msg.channel) {
+        // Step 1: Count total records
+        const totalRes = await this.pg.query('SELECT COUNT(*) FROM org_agents');
+        const total = Number(totalRes.rows[0].count);
+
+        // 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) || [];
+
+        const emailDto = {
+          emailFrom: process.env.PUBLIC_PLATFORM_SUPPORT_EMAIL,
+          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>`
+        };
+
+        try {
+          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 ?? 'Some error occurred while sending prisma ledgerId alert email');
+        }
+      }
+    });
+  }
+
+  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.service.ts

@@ -4,60 +4,84 @@ import { UtilitiesRepository } from './utilities.repository';
 import { AwsService } from '@credebl/aws';
 import { S3 } from 'aws-sdk';
 import { v4 as uuidv4 } from 'uuid';
+import { EmailService } from '@credebl/common/email.service';
+import { EmailDto } from '@credebl/common/dtos/email.dto';
 
 @Injectable()
 export class UtilitiesService {
-    constructor(
-        private readonly logger: Logger,
-        private readonly utilitiesRepository: UtilitiesRepository,
-        private readonly awsService: AwsService
-    ) { }
-
-    async createAndStoreShorteningUrl(payload): Promise<string> {
-        try {
-            const { credentialId, schemaId, credDefId, invitationUrl, attributes } = payload;
-            const invitationPayload = {
-                referenceId: credentialId,
-                invitationPayload: {
-                    schemaId, 
-                    credDefId,
-                    invitationUrl, 
-                    attributes
-                }
-            };
-            await this.utilitiesRepository.saveShorteningUrl(invitationPayload);
-            return `${process.env.API_GATEWAY_PROTOCOL}://${process.env.API_ENDPOINT}/invitation/qr-code/${credentialId}`;
-        } catch (error) {
-            this.logger.error(`[createAndStoreShorteningUrl] - error in create shortening url: ${JSON.stringify(error)}`);
-            throw new RpcException(error);
+  private lastAlertTime: number | null = null;
+
+  constructor(
+    private readonly logger: Logger,
+    private readonly utilitiesRepository: UtilitiesRepository,
+    private readonly awsService: AwsService,
+    private readonly emailService: EmailService
+  ) {}
+
+  async createAndStoreShorteningUrl(payload): Promise<string> {
+    try {
+      const { credentialId, schemaId, credDefId, invitationUrl, attributes } = payload;
+      const invitationPayload = {
+        referenceId: credentialId,
+        invitationPayload: {
+          schemaId,
+          credDefId,
+          invitationUrl,
+          attributes
         }
+      };
+      await this.utilitiesRepository.saveShorteningUrl(invitationPayload);
+      return `${process.env.API_GATEWAY_PROTOCOL}://${process.env.API_ENDPOINT}/invitation/qr-code/${credentialId}`;
+    } catch (error) {
+      this.logger.error(`[createAndStoreShorteningUrl] - error in create shortening url: ${JSON.stringify(error)}`);
+      throw new RpcException(error);
     }
+  }
 
-    async getShorteningUrl(referenceId: string): Promise<object> {
-        try {
-            const getShorteningUrl = await this.utilitiesRepository.getShorteningUrl(referenceId);
-
-            const getInvitationUrl  = {
-                referenceId: getShorteningUrl.referenceId,
-                invitationPayload: getShorteningUrl.invitationPayload
-            };
-
-            return getInvitationUrl;
-        } catch (error) {
-            this.logger.error(`[getShorteningUrl] - error in get shortening url: ${JSON.stringify(error)}`);
-            throw new RpcException(error);
-        }
+  async getShorteningUrl(referenceId: string): Promise<object> {
+    try {
+      const getShorteningUrl = await this.utilitiesRepository.getShorteningUrl(referenceId);
+
+      const getInvitationUrl = {
+        referenceId: getShorteningUrl.referenceId,
+        invitationPayload: getShorteningUrl.invitationPayload
+      };
+
+      return getInvitationUrl;
+    } catch (error) {
+      this.logger.error(`[getShorteningUrl] - error in get shortening url: ${JSON.stringify(error)}`);
+      throw new RpcException(error);
     }
+  }
 
-    async storeObject(payload: {persistent: boolean, storeObj: unknown}): Promise<string> {
-        try {
-            const uuid = uuidv4();
-            const uploadResult:S3.ManagedUpload.SendData = await this.awsService.storeObject(payload.persistent, uuid, payload.storeObj);
-            const url: string = `${process.env.SHORTENED_URL_DOMAIN}/${uploadResult.Key}`;
-            return url;
-        } catch (error) {
-            this.logger.error(error);
-            throw new Error('An error occurred while uploading data to S3. Error::::::');
-        }
+  async storeObject(payload: { persistent: boolean; storeObj: unknown }): Promise<string> {
+    try {
+      const uuid = uuidv4();
+      const uploadResult: S3.ManagedUpload.SendData = await this.awsService.storeObject(
+        payload.persistent,
+        uuid,
+        payload.storeObj
+      );
+      const url: string = `${process.env.SHORTENED_URL_DOMAIN}/${uploadResult.Key}`;
+      return url;
+    } catch (error) {
+      this.logger.error(error);
+      throw new Error('An error occurred while uploading data to S3. Error::::::');
+    }
+  }
+
+  async handleLedgerAlert(emailDto: EmailDto): Promise<void> {
+    // Avoid spamming: send only once every 2 hours
+    const now = Date.now();
+    if (this.lastAlertTime && now - this.lastAlertTime < 2 * 60 * 60 * 1000) {
+      this.logger.log(`ALERT EMAIL ALREADY SENT at ${this.lastAlertTime}, ledgerId WAS SET TO NULL`);
+      return;
     }
+    this.lastAlertTime = now;
+
+    // Send Email
+    await this.emailService.sendEmail(emailDto);
+
+    this.logger.log('ALERT EMAIL SENT, ledgerId WAS SET TO NULL');
+  }
 }

libs/common/src/dtos/email.dto.ts

@@ -1,17 +1,17 @@
 export class EmailDto {
-    emailFrom: string;
-    emailTo: string;
-    emailSubject: string;
-    emailText: string;
-    emailHtml: string;
-    emailAttachments?: AttachmentJSON[];
+  emailFrom: string;
+  emailTo: string | string[];
+  emailSubject: string;
+  emailText: string;
+  emailHtml: string;
+  emailAttachments?: AttachmentJSON[];
 }
 
 interface AttachmentJSON {
-    content: string;
-    filename: string;
-    contentType: string;
-    type?: string;
-    disposition?: string;
-    content_id?: string;
-  }
+  content: string;
+  filename: string;
+  contentType: string;
+  type?: string;
+  disposition?: string;
+  content_id?: string;
+}

libs/common/src/resend-helper-file.ts

@@ -10,7 +10,7 @@ const apiKey = process.env.RESEND_API_KEY;
 if (!apiKey) {
   throw new Error('Missing RESEND_API_KEY in environment variables.');
 }
-const resend = new Resend(process.env.RESEND_API_KEY);
+const resend = new Resend(apiKey);
 
 export const sendWithResend = async (emailDto: EmailDto): Promise<boolean> => {
   try {

libs/org-roles/src/org-roles.module.ts

@@ -1,11 +1,12 @@
-import { PrismaService } from '@credebl/prisma-service';
+import { PrismaServiceModule } from '@credebl/prisma-service';
 import { Logger } from '@nestjs/common';
 import { Module } from '@nestjs/common';
 import { OrgRolesRepository } from '../repositories';
 import { OrgRolesService } from './org-roles.service';
 
 @Module({
-  providers: [OrgRolesService, OrgRolesRepository, Logger, PrismaService],
+  imports: [PrismaServiceModule],
+  providers: [OrgRolesService, OrgRolesRepository, Logger],
   exports: [OrgRolesService]
 })
 export class OrgRolesModule {}

libs/prisma-service/prisma/migrations/20251125062851_add_ledger_null_trigger/migration.sql

@@ -0,0 +1,21 @@
+-- Create the function
+CREATE OR REPLACE FUNCTION alert_ledger_null()
+RETURNS trigger AS $$
+BEGIN
+    IF NEW."ledgerId" IS NULL THEN
+        PERFORM pg_notify('ledger_null', json_build_object(
+            'agentId', NEW.id,
+            'orgId', NEW."orgId",
+            'timestamp', now()
+        )::text);
+    END IF;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Create the trigger
+CREATE TRIGGER ledger_null_trigger
+AFTER UPDATE ON org_agents
+FOR EACH ROW
+WHEN (NEW."ledgerId" IS NULL AND OLD."ledgerId" IS NOT NULL)
+EXECUTE FUNCTION alert_ledger_null();