fix: added auth to register holder notification api

.env.demo

@@ -26,6 +26,9 @@ NATS_HOST=your-ip
 NATS_PORT=4222
 NATS_URL=nats://your-ip:4222
 
+NATS_PASSWORD=xxxx
+NATS_USER=xxxx
+
 REDIS_HOST=your-ip
 REDIS_PORT=6379
 
@@ -230,3 +233,16 @@ PRISMA_LOGS = error
 
 # DB_ALERT_ENABLE=
 # DB_ALERT_EMAILS=
+
+
+
+CLIENT_EMAIL=
+PRIVATE_KEY=
+PROJECT_ID=
+
+CONSUMER_CONFIG_ACK_WAIT=10_000  # IN nanos(10_000)
+CONSUMER_CONFIG_MAX_DELIVER=4
+
+AGGREGATE_STREAM=aggregate
+DID_STREAM=did-notify
+PULL_CONSUMER=hub-pull-consumer

.env.sample

@@ -59,6 +59,9 @@ NATS_HOST='0.0.0.0'
 NATS_PORT=4222
 NATS_URL=nats://0.0.0.0:4222
 
+NATS_PASSWORD=xxxx
+NATS_USER=xxxx
+
 REDIS_HOST='0.0.0.0'
 REDIS_PORT=6379
 
@@ -259,3 +262,15 @@ RESEND_API_KEY=re_xxxxxxxxxx
 # DB_ALERT_EMAILS=
 # Boolean: to enable/disable db alerts. This needs the 'utility' microservice
 # DB_ALERT_ENABLE=
+
+
+CLIENT_EMAIL=
+PRIVATE_KEY=
+PROJECT_ID=
+
+CONSUMER_CONFIG_ACK_WAIT=10_000  # IN nanos(10_000)
+CONSUMER_CONFIG_MAX_DELIVER=4
+
+AGGREGATE_STREAM=aggregate
+DID_STREAM=did-notify
+PULL_CONSUMER=hub-pull-consumer

apps/api-gateway/src/authz/guards/client-access-guard.ts

@@ -0,0 +1,17 @@
+import { Injectable, CanActivate, ExecutionContext, UnauthorizedException } from '@nestjs/common';
+import { Observable } from 'rxjs';
+
+@Injectable()
+export class ClientAccessGuard implements CanActivate {
+  canActivate(context: ExecutionContext): boolean | Promise<boolean> | Observable<boolean> {
+    const request = context.switchToHttp().getRequest();
+
+    const { user } = request;
+
+    if (!user || !Object.prototype.hasOwnProperty.call(user, 'client_id')) {
+      throw new UnauthorizedException('You do not have access');
+    }
+
+    return true;
+  }
+}

apps/api-gateway/src/main.ts

@@ -125,6 +125,7 @@ async function bootstrap(): Promise<void> {
       xssFilter: true
     })
   );
+  Logger.log('API-Gateway is listening to NATS', 'NATS-CONNECTION');
   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}`, 'Success');
 

apps/api-gateway/src/notification/notification.controller.ts

@@ -1,6 +1,7 @@
 import { CustomExceptionFilter } from '@credebl/common/exception-handler';
-import { Body, Controller, HttpStatus, Logger, Post, Res, UseFilters } from '@nestjs/common';
+import { Body, Controller, HttpStatus, Logger, Post, Res, UseFilters, UseGuards } from '@nestjs/common';
 import {
+  ApiBearerAuth,
   ApiExcludeEndpoint,
   ApiForbiddenResponse,
   ApiOperation,
@@ -20,6 +21,8 @@ import { IResponse } from '@credebl/common/interfaces/response.interface';
 import { Response } from 'express';
 import { ResponseMessages } from '@credebl/common/response-messages';
 import { NotificationService } from './notification.service';
+import { ClientAccessGuard } from '../authz/guards/client-access-guard';
+import { AuthGuard } from '@nestjs/passport';
 
 @Controller('notification')
 @UseFilters(CustomExceptionFilter)
@@ -93,6 +96,8 @@ export class NotificationController {
    */
   @Post('/register/holder-notification')
   // @ApiExcludeEndpoint()
+  @ApiBearerAuth()
+  @UseGuards(AuthGuard('jwt'), ClientAccessGuard)
   @ApiOperation({
     summary: `Register holder for notification`,
     description: `Register holder for notification`

apps/notification/package.json

@@ -0,0 +1,9 @@
+{
+  "name": "@credebl/notification",
+  "version": "1.0.0",
+  "private": true,
+  "type": "module",
+  "dependencies": {
+    "firebase-admin": "^13.6.0"
+  }
+}

apps/notification/src/nats/interfaces.ts

@@ -0,0 +1,14 @@
+export interface INatsEventMessage {
+  event: string;
+  sessionId: string;
+  timestamp: string;
+  senderCode: string;
+}
+
+export interface INatsUserRequestData {
+  did: string;
+  sessions: {
+    sessionId: string;
+    orgCode: string;
+  }[];
+}

apps/notification/src/nats/jetstream.consumer.ts

@@ -0,0 +1,164 @@
+import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common';
+import { Consumer } from 'nats';
+import { NatsService } from './nats.service';
+import {
+  DID_STREAM,
+  ensureConsumer,
+  ensureDidStream,
+  publishToJetStream,
+  PULL_CONSUMER,
+  AGGREGATE_STREAM
+} from './jetstream.setup';
+import { PendingAckStore } from './pendingAckStore';
+import { HolderNotificationRepository } from '../holder-notification.repository';
+import { Message } from 'firebase-admin/lib/messaging/messaging-api';
+import * as admin from 'firebase-admin';
+import { IHolderNotification } from '@credebl/common/interfaces/holder-notification.interfaces';
+
+const EVENT_PRESENTATION_ACK = 'presentation.ack';
+const EVENT_PRESENTATION_PURGED = 'presentation.purged';
+@Injectable()
+export class JetStreamConsumer implements OnApplicationBootstrap {
+  constructor(
+    private readonly nats: NatsService,
+    private readonly logger: Logger,
+    private readonly pendingAckStore: PendingAckStore,
+    private readonly holderNotificationRepository: HolderNotificationRepository
+  ) {}
+
+  async onApplicationBootstrap(): Promise<void> {
+    if (!admin.apps.length) {
+      const projectId = process.env.PROJECT_ID;
+      const clientEmail = process.env.CLIENT_EMAIL;
+      const privateKey = process.env.PRIVATE_KEY;
+
+      if (!projectId || !clientEmail || !privateKey) {
+        throw new Error('Missing Firebase credentials: PROJECT_ID, CLIENT_EMAIL, and PRIVATE_KEY are required');
+      }
+      admin.initializeApp({
+        credential: admin.credential.cert({
+          projectId,
+          clientEmail,
+          privateKey: privateKey.replace(/\\n/g, '\n')
+        })
+      });
+    }
+
+    const js = this.nats.jetstream(); // ✅ now safe
+
+    const jsm = this.nats.jetstreamManager();
+    await ensureConsumer(jsm);
+    const consumer: Consumer = await js.consumers.get(AGGREGATE_STREAM, PULL_CONSUMER);
+
+    this.logger.log('[NATS] JetStream consumer started');
+
+    this.consume(consumer).catch((err) => {
+      this.logger.error('[NATS] Consumer crashed', err);
+    });
+  }
+
+  private async consume(consumer: Consumer): Promise<void> {
+    this.logger.log(`[NATS] Starting to consume messages from consumer ${consumer.info}`);
+    for await (const msg of await consumer.consume()) {
+      try {
+        const { subject } = msg;
+        this.logger.log(`[NATS] Message subject: ${subject}`);
+        // eslint-disable-next-line @typescript-eslint/no-unused-vars
+        const [_, domain, event, orgCode, sessionId] = subject.split('.');
+        const consumerName = `notify-session-${sessionId}`;
+
+        this.logger.log({
+          domain,
+          event,
+          orgCode,
+          sessionId
+        });
+
+        const payload = msg.json();
+        const notificationDetail = await this.holderNotificationRepository.getHolderNotificationBySessionId(sessionId);
+        this.logger.debug(`[NATS] Message received ${JSON.stringify(payload)}`);
+        this.logger.log(`[NATS] Processing message, ${JSON.stringify({ deliveryCount: msg.info.deliveryCount })}`);
+        const maxDeliver = process.env.CONSUMER_CONFIG_MAX_DELIVER
+          ? Number(process.env.CONSUMER_CONFIG_MAX_DELIVER)
+          : 4;
+        if (maxDeliver - 1 < msg.info.deliveryCount) {
+          //------------ Moving message to DID stream ---------------//
+          await this.moveMsgToDidStream(notificationDetail, msg);
+
+          //------------- Sending Push Notification via FCM ----------------//
+          this.sendNotificationToHolder(event, orgCode, notificationDetail);
+          msg.ack();
+        } else {
+          // ------------- Publishing Messages via NATS ----------------//
+
+          const ackKey = this.pendingAckStore.save(AGGREGATE_STREAM, consumerName, msg);
+
+          this.logger.log(`[NATS] Notification detail fetched for session ID: ${sessionId}`);
+          if (!notificationDetail) {
+            this.logger.error(`[NATS] No notification detail found for session ID: ${sessionId}`);
+            msg.nak();
+            continue;
+          }
+          await this.nats.publish(`${notificationDetail.holderDid}`, {
+            payload,
+            ackKey,
+            subject: msg.subject,
+            event: `${domain}.${event}`
+          });
+
+          this.logger.log(`[NATS] Message published to ${notificationDetail.holderDid} for session ${sessionId}`);
+        }
+
+        // business logic
+        // msg.ack();
+      } catch (err) {
+        this.logger.error('[NATS] Processing failed', err);
+        msg.nak();
+      }
+    }
+  }
+
+  private sendNotificationToHolder(event: string, orgCode: string, notificationDetail: IHolderNotification): void {
+    let notificationTitle = '';
+    if (EVENT_PRESENTATION_ACK === event) {
+      notificationTitle = `Your data delivered to ${orgCode}`;
+    } else if (EVENT_PRESENTATION_PURGED === event) {
+      notificationTitle = `Your data purged from ${orgCode}`;
+    }
+    this.logger.log('Now push notifications will be sent to user');
+    const notificationPayload: Message = {
+      notification: {
+        title: notificationTitle,
+        body: `Open an app to view more details`
+      },
+      token: notificationDetail.fcmToken
+    };
+    // Send the notification to the specified device
+    admin
+      .messaging()
+      .send(notificationPayload)
+      .then((response) => {
+        this.logger.log('Successfully sent message:', response);
+      })
+      .catch((error) => {
+        this.logger.error('Error sending message:', error);
+      });
+  }
+
+  private async moveMsgToDidStream(notificationDetail: IHolderNotification, msg): Promise<void> {
+    this.logger.log('[NATS] Moving message to DID stream for FCM notification');
+    //const notifyStream = process.env.STREAM_NOTIFY ?? "notify";
+    const subjectName = `${DID_STREAM}.${notificationDetail.holderDid}`;
+    const jsm = this.nats.jetstreamManager();
+    const jsc = this.nats.jetstream();
+    await ensureDidStream(jsm);
+
+    const decoder = new TextDecoder(); // or 'UTF-8', 'windows-1251', etc.
+    const msgData = decoder.decode(msg.data);
+
+    //await ensureStreamExists(streamName);
+    // Publish to JetStream for guaranteed delivery
+    const jsAck = await publishToJetStream(subjectName, msgData, jsc);
+    this.logger.log(`[NATS] Message moved to DID stream: ${JSON.stringify(jsAck)}`);
+  }
+}