From 3bfe9da5819f2aa8ae511123af2c1f5e3a8da769 Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Tue, 13 Jan 2026 18:04:47 +0530 Subject: [PATCH 01/13] fix: added auth to register holder notification api Signed-off-by: Rinkal Bhojani --- .../src/authz/guards/client-access-guard.ts | 17 +++++++++++++++++ .../src/notification/notification.controller.ts | 7 ++++++- apps/notification/src/notification.service.ts | 6 +++--- 3 files changed, 26 insertions(+), 4 deletions(-) create mode 100644 apps/api-gateway/src/authz/guards/client-access-guard.ts diff --git a/apps/api-gateway/src/authz/guards/client-access-guard.ts b/apps/api-gateway/src/authz/guards/client-access-guard.ts new file mode 100644 index 000000000..de9390cbf --- /dev/null +++ b/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 | Observable { + const request = context.switchToHttp().getRequest(); + + const { user } = request; + + if (!user.hasOwnProperty('client_id')) { + throw new UnauthorizedException('You do not have access'); + } + + return true; + } +} diff --git a/apps/api-gateway/src/notification/notification.controller.ts b/apps/api-gateway/src/notification/notification.controller.ts index e962d2e3f..140c18504 100644 --- a/apps/api-gateway/src/notification/notification.controller.ts +++ b/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` diff --git a/apps/notification/src/notification.service.ts b/apps/notification/src/notification.service.ts index dd44defae..973de0346 100644 --- a/apps/notification/src/notification.service.ts +++ b/apps/notification/src/notification.service.ts @@ -4,12 +4,12 @@ import { RpcException } from '@nestjs/microservices'; import { NotificationRepository } from './notification.repository'; import { ResponseMessages } from '@credebl/common/response-messages'; import { CommonService } from '@credebl/common'; +import { HolderNotificationRepository } from './holder-notification.repository'; +import { NotificationStatus } from '@credebl/enum/enum'; import { ICreateHolderNotification, IHolderNotification -} from '../../../libs/common/src/interfaces/holder-notification.interfaces'; -import { HolderNotificationRepository } from './holder-notification.repository'; -import { NotificationStatus } from '@credebl/enum/enum'; +} from '@credebl/common/interfaces/holder-notification.interfaces'; @Injectable() export class NotificationService { From 538b5e3b76be5b46d0e06542725173d4c5513ce5 Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Thu, 22 Jan 2026 12:05:40 +0530 Subject: [PATCH 02/13] feat: Aggregating leaf stream and consuming messages Signed-off-by: Rinkal Bhojani --- .env.demo | 3 + .env.sample | 3 + .../src/nats/jetstream.consumer.ts | 61 +++++++++++++++++ apps/notification/src/nats/jetstream.setup.ts | 65 +++++++++++++++++++ apps/notification/src/nats/nats-subscriber.ts | 53 +++++++++++++++ apps/notification/src/nats/nats.module.ts | 26 ++++++++ apps/notification/src/nats/nats.service.ts | 57 ++++++++++++++++ apps/notification/src/notification.module.ts | 4 +- .../src/oid4vc-verification.service.ts | 4 +- 9 files changed, 273 insertions(+), 3 deletions(-) create mode 100644 apps/notification/src/nats/jetstream.consumer.ts create mode 100644 apps/notification/src/nats/jetstream.setup.ts create mode 100644 apps/notification/src/nats/nats-subscriber.ts create mode 100644 apps/notification/src/nats/nats.module.ts create mode 100644 apps/notification/src/nats/nats.service.ts diff --git a/.env.demo b/.env.demo index 7c5a134e1..4b8032d5e 100644 --- a/.env.demo +++ b/.env.demo @@ -26,6 +26,9 @@ NATS_HOST=your-ip NATS_PORT=4222 NATS_URL=nats://your-ip:4222 +NATS_USER=xxxx +NATS_PASSWORD=xxxx + REDIS_HOST=your-ip REDIS_PORT=6379 diff --git a/.env.sample b/.env.sample index 7cc321598..9eedf47d2 100644 --- a/.env.sample +++ b/.env.sample @@ -59,6 +59,9 @@ NATS_HOST='0.0.0.0' NATS_PORT=4222 NATS_URL=nats://0.0.0.0:4222 +NATS_USER=xxxx +NATS_PASSWORD=xxxx + REDIS_HOST='0.0.0.0' REDIS_PORT=6379 diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts new file mode 100644 index 000000000..88f3fc6e1 --- /dev/null +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -0,0 +1,61 @@ +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; +import { Consumer } from 'nats'; +import { NatsService } from './nats.service'; +import { CONSUMER, STREAM } from './jetstream.setup'; + +@Injectable() +export class JetStreamConsumer implements OnApplicationBootstrap { + constructor( + private readonly nats: NatsService, + private readonly logger: Logger + ) {} + + async onApplicationBootstrap(): Promise { + const js = this.nats.jetstream(); // ✅ now safe + + const consumer: Consumer = await js.consumers.get(STREAM, CONSUMER); + // const consumer: Consumer = await js.consumers.get( + // STREAM, + // { + // // name_prefix: CONSUMER, + // filterSubjects: ['hubStream.presentation.test.blr'] + // } + // ); + + 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 { + 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('.'); + + this.logger.log({ + domain, + event, + orgCode, + sessionId + }); + + const payload = msg.json(); + this.logger.log('[NATS] Message received', payload); + + await this.nats.publish('did-key', payload); // ACK reply to APP + this.logger.log('[NATS] Message published to did:key:z6Mkm4VPWQRRSJSgoPVQ2oB8EfRqDGDQJBudpqppovJYeLpu'); + + // business logic + msg.ack(); + } catch (err) { + this.logger.error('[NATS] Processing failed', err); + msg.nak(); + } + } + } +} diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts new file mode 100644 index 000000000..d3e4b752f --- /dev/null +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -0,0 +1,65 @@ +import { Logger } from '@nestjs/common'; +import { AckPolicy, DeliverPolicy, JetStreamManager, RetentionPolicy, StorageType, nanos } from 'nats'; + +// export const STREAM = 'airport-events'; +// export const SUBJECTS = ['airport.events.*']; +// export const CONSUMER = 'hub-consumer'; + +// export const STREAM = 'hubStream'; +// export const SUBJECTS = ['hubStream.presentation.*', 'hubStream.consent.*']; +// export const CONSUMER = 'hub-consumer'; + +export const SOURCE_STREAM = 'notify'; +export const STREAM = 'aggregate'; +export const SUBJECTS = ['aggregate.>']; +export const CONSUMER = 'hub-consumer'; + +const logger = new Logger('JetStreamSetup'); +/** + * Ensures the JetStream stream exists + */ +export async function ensureStream(jsm: JetStreamManager): Promise { + try { + // await jsm.streams.delete(STREAM); + await jsm.streams.info(STREAM); + logger.log(`[NATS] Stream '${STREAM}' already exists`); + } catch { + logger.log(`[NATS] Creating stream '${STREAM}'`); + await jsm.streams.add({ + name: STREAM, + //subjects: SUBJECTS, + retention: RetentionPolicy.Workqueue, + storage: StorageType.File, + sources: [ + { name: SOURCE_STREAM, domain: 'del' }, + { name: SOURCE_STREAM, domain: 'blr' } + ] + }); + } +} + +/** + * Ensures the JetStream consumer exists + */ +export async function ensureConsumer(jsm: JetStreamManager): Promise { + try { + // await jsm.consumers.delete(STREAM, CONSUMER); + // await jsm.consumers.update(STREAM, CONSUMER, { + // // eslint-disable-next-line camelcase + // // filter_subject: '*.presentation.test.blr' + // }); + await jsm.consumers.info(STREAM, CONSUMER); + logger.log(`[NATS] Consumer '${CONSUMER}' already exists`); + } catch { + logger.log(`[NATS] Creating consumer '${CONSUMER}'`); + await jsm.consumers.add(STREAM, { + name: CONSUMER, + durable_name: CONSUMER, + ack_policy: AckPolicy.Explicit, + deliver_policy: DeliverPolicy.All, + ack_wait: nanos(30_000), + max_deliver: 5 + // ,filter_subjects: ['hubStream.presentation.test.*'] + }); + } +} diff --git a/apps/notification/src/nats/nats-subscriber.ts b/apps/notification/src/nats/nats-subscriber.ts new file mode 100644 index 000000000..d5850da62 --- /dev/null +++ b/apps/notification/src/nats/nats-subscriber.ts @@ -0,0 +1,53 @@ +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; +import { NatsService } from './nats.service'; + +const sd = new TextDecoder(); +@Injectable() +export class TestSubsciber implements OnApplicationBootstrap { + constructor( + private readonly nats: NatsService, + private readonly logger: Logger + ) {} + + async onApplicationBootstrap(): Promise { + // this.subscribe('activate.user', async (data) => { + // this.logger.log('[NATS] TestSubsciber received data:', data); + // }); + } + + private async subscribe(subject: string, handler: (data: unknown) => void | Promise): Promise { + try { + this.logger.log('[NATS] Setting up subscription', { + subject + }); + const nc = await this.nats.nc; + const subscription = nc.subscribe(subject); + + // eslint-disable-next-line @typescript-eslint/explicit-function-return-type + (async () => { + for await (const msg of subscription) { + try { + const rawData = sd.decode(msg.data); + const data = JSON.parse(rawData); + await handler(data); + } catch (error) { + this.logger.error('[NATS] Error processing subscribed message', { + subject, + error: error instanceof Error ? error.message : 'Unknown error' + }); + } + } + })(); + + this.logger.log('[NATS] Subscription active', { + subject + }); + } catch (error) { + this.logger.error('[NATS] Failed to set up subscription', { + subject, + error: error instanceof Error ? error.message : 'Unknown error' + }); + throw error; + } + } +} diff --git a/apps/notification/src/nats/nats.module.ts b/apps/notification/src/nats/nats.module.ts new file mode 100644 index 000000000..60c82952f --- /dev/null +++ b/apps/notification/src/nats/nats.module.ts @@ -0,0 +1,26 @@ +import { Logger, Module, OnModuleInit } from '@nestjs/common'; +import { NatsService } from './nats.service'; +import { JetStreamConsumer } from './jetstream.consumer'; +import { ensureConsumer, ensureStream } from './jetstream.setup'; +import { TestSubsciber } from './nats-subscriber'; + +@Module({ + providers: [NatsService, JetStreamConsumer, TestSubsciber], + exports: [NatsService] +}) +export class NatsModule implements OnModuleInit { + constructor( + private readonly nats: NatsService, + private readonly logger: Logger + ) {} + + async onModuleInit(): Promise { + this.logger.log('[NATS] Module initializing...'); + await this.nats.connect(); + + const jsm = this.nats.jetstreamManager(); + + await ensureStream(jsm); + await ensureConsumer(jsm); + } +} diff --git a/apps/notification/src/nats/nats.service.ts b/apps/notification/src/nats/nats.service.ts new file mode 100644 index 000000000..bb1a2009e --- /dev/null +++ b/apps/notification/src/nats/nats.service.ts @@ -0,0 +1,57 @@ +import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common'; +import { connect, JetStreamClient, JetStreamManager, NatsConnection, usernamePasswordAuthenticator } from 'nats'; + +@Injectable() +export class NatsService implements OnModuleDestroy { + public nc!: NatsConnection; + private js!: JetStreamClient; + private jsm!: JetStreamManager; + private connected = false; + private readonly logger = new Logger('NatsService'); + + async connect(): Promise { + if (this.connected) { + return; + } + + this.logger.log('[NATS] starting connection...'); + + this.nc = await connect({ + servers: [`${process.env.NATS_HOST}:${process.env.NATS_PORT}`], + authenticator: usernamePasswordAuthenticator(`${process.env.NATS_USER}`, `${process.env.NATS_PASSWORD}`) + }); + this.js = this.nc.jetstream(); + this.jsm = await this.nc.jetstreamManager(); + + this.connected = true; + this.logger.log('[NATS] connection established'); + return this.nc; + } + + jetstream(): JetStreamClient { + if (!this.connected) { + throw new Error('NATS not connected yet'); + } + return this.js; + } + + jetstreamManager(): JetStreamManager { + if (!this.connected) { + throw new Error('NATS not connected yet'); + } + return this.jsm; + } + + async onModuleDestroy(): Promise { + if (this.nc) { + await this.nc.drain(); + } + } + + async publish(subject: string, payload: unknown): Promise { + if (!this.connected) { + throw new Error('NATS not connected yet'); + } + this.nc.publish(subject, Buffer.from(JSON.stringify(payload)), { reply: 'ack-notification' }); + } +} diff --git a/apps/notification/src/notification.module.ts b/apps/notification/src/notification.module.ts index 8da1884a1..97bc31fc0 100644 --- a/apps/notification/src/notification.module.ts +++ b/apps/notification/src/notification.module.ts @@ -14,6 +14,7 @@ import { ConfigModule as PlatformConfig } from '@credebl/config/config.module'; import { LoggerModule } from '@credebl/logger/logger.module'; import { ContextInterceptorModule } from '@credebl/context/contextInterceptorModule'; import { HolderNotificationRepository } from './holder-notification.repository'; +import { NatsModule } from './nats/nats.module'; @Module({ imports: [ @@ -30,7 +31,8 @@ import { HolderNotificationRepository } from './holder-notification.repository'; LoggerModule, PlatformConfig, ContextInterceptorModule, - CacheModule.register({ host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }) + CacheModule.register({ host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }), + NatsModule ], controllers: [NotificationController], providers: [NotificationService, NotificationRepository, HolderNotificationRepository, PrismaService, Logger] diff --git a/apps/oid4vc-verification/src/oid4vc-verification.service.ts b/apps/oid4vc-verification/src/oid4vc-verification.service.ts index ff6a83df4..6ff6d4ffb 100644 --- a/apps/oid4vc-verification/src/oid4vc-verification.service.ts +++ b/apps/oid4vc-verification/src/oid4vc-verification.service.ts @@ -393,7 +393,7 @@ export class Oid4vpVerificationService extends BaseService { return createdSession; } catch (error) { this.logger.error( - `[createVerificationPresentation] - error creating verification presentation: ${JSON.stringify(error?.response ?? error)}` + `[createVerificationPresentation] - error creating verification presentation: ${JSON.stringify(error)}` ); throw new RpcException(error?.response ?? error); } @@ -451,7 +451,7 @@ export class Oid4vpVerificationService extends BaseService { this.logger.debug(`[getVerificationSessionResponse] response fetched successfully for orgId=${orgId}`); return verifiers; } catch (error) { - this.logger.error(`[getVerificationSessionResponse] - error: ${JSON.stringify(error?.response ?? error)}`); + this.logger.error(`[getVerificationSessionResponse] - error: ${JSON.stringify(error)}`); throw new RpcException(error?.response ?? error); } } From f6e00b61f6a2f469904f6c93cd0e77154f96cf69 Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Thu, 22 Jan 2026 12:09:19 +0530 Subject: [PATCH 03/13] fix:removed unwanted commented code Signed-off-by: Rinkal Bhojani --- apps/notification/src/nats/jetstream.setup.ts | 8 -------- 1 file changed, 8 deletions(-) diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index d3e4b752f..a80d15731 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -1,14 +1,6 @@ import { Logger } from '@nestjs/common'; import { AckPolicy, DeliverPolicy, JetStreamManager, RetentionPolicy, StorageType, nanos } from 'nats'; -// export const STREAM = 'airport-events'; -// export const SUBJECTS = ['airport.events.*']; -// export const CONSUMER = 'hub-consumer'; - -// export const STREAM = 'hubStream'; -// export const SUBJECTS = ['hubStream.presentation.*', 'hubStream.consent.*']; -// export const CONSUMER = 'hub-consumer'; - export const SOURCE_STREAM = 'notify'; export const STREAM = 'aggregate'; export const SUBJECTS = ['aggregate.>']; From 917f42fc65143eb4c43e55da0cfe10bf62d6322f Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Tue, 27 Jan 2026 16:37:16 +0530 Subject: [PATCH 04/13] feat: logic to send nats based notification message once user is active Signed-off-by: Rinkal Bhojani --- apps/notification/src/nats/interfaces.ts | 14 +++ .../src/nats/jetstream.consumer.ts | 11 +- apps/notification/src/nats/jetstream.setup.ts | 30 ++++- apps/notification/src/nats/nats-subscriber.ts | 104 +++++++++++++++--- apps/notification/src/nats/nats.module.ts | 24 +++- apps/notification/src/nats/nats.service.ts | 3 +- apps/notification/src/nats/pendingAckStore.ts | 38 +++++++ 7 files changed, 198 insertions(+), 26 deletions(-) create mode 100644 apps/notification/src/nats/interfaces.ts create mode 100644 apps/notification/src/nats/pendingAckStore.ts diff --git a/apps/notification/src/nats/interfaces.ts b/apps/notification/src/nats/interfaces.ts new file mode 100644 index 000000000..a528a4f03 --- /dev/null +++ b/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; + }[]; +} diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index 88f3fc6e1..4952c68c0 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -1,7 +1,6 @@ import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { Consumer } from 'nats'; import { NatsService } from './nats.service'; -import { CONSUMER, STREAM } from './jetstream.setup'; @Injectable() export class JetStreamConsumer implements OnApplicationBootstrap { @@ -11,9 +10,9 @@ export class JetStreamConsumer implements OnApplicationBootstrap { ) {} async onApplicationBootstrap(): Promise { - const js = this.nats.jetstream(); // ✅ now safe + // const js = this.nats.jetstream(); // ✅ now safe - const consumer: Consumer = await js.consumers.get(STREAM, CONSUMER); + // const consumer: Consumer = await js.consumers.get(STREAM, CONSUMER); // const consumer: Consumer = await js.consumers.get( // STREAM, // { @@ -24,9 +23,9 @@ export class JetStreamConsumer implements OnApplicationBootstrap { this.logger.log('[NATS] JetStream consumer started'); - this.consume(consumer).catch((err) => { - this.logger.error('[NATS] Consumer crashed', err); - }); + // this.consume(consumer).catch((err) => { + // this.logger.error('[NATS] Consumer crashed', err); + // }); } private async consume(consumer: Consumer): Promise { diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index a80d15731..b1060e7ee 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -1,5 +1,6 @@ +/* eslint-disable camelcase */ import { Logger } from '@nestjs/common'; -import { AckPolicy, DeliverPolicy, JetStreamManager, RetentionPolicy, StorageType, nanos } from 'nats'; +import { AckPolicy, DeliverPolicy, JetStreamManager, ReplayPolicy, RetentionPolicy, StorageType, nanos } from 'nats'; export const SOURCE_STREAM = 'notify'; export const STREAM = 'aggregate'; @@ -55,3 +56,30 @@ export async function ensureConsumer(jsm: JetStreamManager): Promise { }); } } + +export async function ensureSessionConsumer(jsm: JetStreamManager, sessionId: string): Promise { + const consumerName = `notify-session-${sessionId}`; + + try { + await jsm.consumers.info(STREAM, consumerName); + logger.log(`[NATS] Consumer '${consumerName}' already exists`); + return consumerName; + } catch (error) { + logger.error( + `Error creatingg consumer info for '${consumerName}': ${error instanceof Error ? error.message : 'Unknown error'}` + ); + logger.log(`[NATS] Creating consumer '${consumerName}'`); + await jsm.consumers.add(STREAM, { + name: consumerName, + // durable_name: consumerName, + ack_policy: AckPolicy.Explicit, + deliver_policy: DeliverPolicy.All, + filter_subject: `notify.*.*.*.${sessionId}`, + replay_policy: ReplayPolicy.Instant, + max_ack_pending: 1000, + inactive_threshold: 30_000 + }); + + return consumerName; + } +} diff --git a/apps/notification/src/nats/nats-subscriber.ts b/apps/notification/src/nats/nats-subscriber.ts index d5850da62..7d290e4ec 100644 --- a/apps/notification/src/nats/nats-subscriber.ts +++ b/apps/notification/src/nats/nats-subscriber.ts @@ -1,25 +1,34 @@ import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { NatsService } from './nats.service'; +import { INatsUserRequestData } from './interfaces'; +import { ensureSessionConsumer, STREAM } from './jetstream.setup'; +import { JetStreamClient } from 'nats'; +import { PendingAckStore } from './pendingAckStore'; const sd = new TextDecoder(); @Injectable() export class TestSubsciber implements OnApplicationBootstrap { constructor( private readonly nats: NatsService, - private readonly logger: Logger + private readonly logger: Logger, + private readonly pendingAckStore: PendingAckStore ) {} async onApplicationBootstrap(): Promise { - // this.subscribe('activate.user', async (data) => { - // this.logger.log('[NATS] TestSubsciber received data:', data); - // }); + this.subscribe('active.user', async (data) => { + this.logger.log(`[NATS] activate.user Subsciber received data: ${JSON.stringify(data)}`); + await this.activateUserConsumers(data as INatsUserRequestData); + }); + + this.subscribe('user.ack', async (data) => { + this.logger.log('[NATS] user.ack Subsciber received data:', data); + await this.handleUserAck(data as { ackKey: string }); + }); } private async subscribe(subject: string, handler: (data: unknown) => void | Promise): Promise { try { - this.logger.log('[NATS] Setting up subscription', { - subject - }); + this.logger.log(`[NATS] Setting up subscription ${JSON.stringify(subject)}`); const nc = await this.nats.nc; const subscription = nc.subscribe(subject); @@ -31,10 +40,12 @@ export class TestSubsciber implements OnApplicationBootstrap { const data = JSON.parse(rawData); await handler(data); } catch (error) { - this.logger.error('[NATS] Error processing subscribed message', { - subject, - error: error instanceof Error ? error.message : 'Unknown error' - }); + this.logger.error( + `[NATS] Error processing subscribed message ${JSON.stringify({ + subject, + error: error instanceof Error ? error.message : 'Unknown error' + })}` + ); } } })(); @@ -43,11 +54,74 @@ export class TestSubsciber implements OnApplicationBootstrap { subject }); } catch (error) { - this.logger.error('[NATS] Failed to set up subscription', { - subject, - error: error instanceof Error ? error.message : 'Unknown error' - }); + this.logger.error( + `[NATS] Failed to set up subscription ${JSON.stringify({ + subject, + error: error instanceof Error ? error.message : 'Unknown error' + })}` + ); throw error; } } + + private async activateUserConsumers(data: INatsUserRequestData): Promise { + const { did, sessions } = data; + + const js = this.nats.jetstream(); + const jsm = this.nats.jetstreamManager(); + this.logger.log(`[NATS] Started activating consumer for user DID: ${did} with sessions count: ${sessions.length}`); + + // Create all consumers in parallel first (dependency) + await Promise.all(sessions.map((session) => ensureSessionConsumer(jsm, session.sessionId))); + + // After all consumers are created, consume messages from all sessions in parallel + await Promise.all(sessions.map((session) => this.consumeSessionMessages(js, session.sessionId, did))); + } + + private async consumeSessionMessages(js: JetStreamClient, sessionId: string, did: string): Promise { + const consumerName = `notify-session-${sessionId}`; + this.logger.log(`[NATS] Getting consumer ${consumerName} to fetch messages for session ${sessionId}`); + const consumer = await js.consumers.get(STREAM, consumerName); + + const iter = await consumer.fetch({ + // eslint-disable-next-line camelcase + max_messages: 10 + }); + this.logger.log(`[NATS] Started fetching messages for consumer ${consumerName}`); + for await (const msg of iter) { + const payload = msg.json(); + // eslint-disable-next-line @typescript-eslint/no-unused-vars + const [_, domain, event, orgCode, sessionId] = msg.subject.split('.'); + + const ackKey = this.pendingAckStore.save('notify', consumerName, msg); + + await this.nats.publish(`${did}`, { + payload, + ackKey, // opaque token user must return + subject: msg.subject, + event: `${domain}.${event}` + }); + this.logger.log(`[NATS] Published message to user DID ${did} for session ${sessionId}`); + } + this.logger.log(`[NATS] Finished processing messages for session ${sessionId}, deleting consumer ${consumerName}`); + await this.nats.jetstreamManager().consumers.delete(STREAM, consumerName); + this.logger.log(`[NATS] Deleted consumer ${consumerName} after processing messages for session ${sessionId}`); + } + + async handleUserAck(data: { ackKey: string }): Promise { + const msg = this.pendingAckStore.get(data.ackKey); + + if (!msg) { + this.logger.warn('[NATS] ACK received but message not found or already acked', data); + return; + } + + try { + msg.ack(); + this.pendingAckStore.delete(data.ackKey); + this.logger.log('[NATS] Message ACKed via user confirmation'); + } catch (err) { + this.logger.error(`[NATS] Failed to ACK message ${JSON.stringify(err)}`); + } + } } diff --git a/apps/notification/src/nats/nats.module.ts b/apps/notification/src/nats/nats.module.ts index 60c82952f..69b7bca81 100644 --- a/apps/notification/src/nats/nats.module.ts +++ b/apps/notification/src/nats/nats.module.ts @@ -1,11 +1,29 @@ import { Logger, Module, OnModuleInit } from '@nestjs/common'; import { NatsService } from './nats.service'; import { JetStreamConsumer } from './jetstream.consumer'; -import { ensureConsumer, ensureStream } from './jetstream.setup'; +import { ensureStream } from './jetstream.setup'; import { TestSubsciber } from './nats-subscriber'; +import { ConfigModule } from '@nestjs/config'; +import { CommonModule } from '@credebl/common'; +import { GlobalConfigModule } from '@credebl/config'; +import { ConfigModule as PlatformConfig } from '@credebl/config/config.module'; +import { LoggerModule } from '@credebl/logger'; +import { ContextInterceptorModule } from '@credebl/context'; +import { CacheModule } from '@nestjs/cache-manager'; +import { PendingAckStore } from './pendingAckStore'; @Module({ - providers: [NatsService, JetStreamConsumer, TestSubsciber], + imports: [ + ConfigModule.forRoot(), + + CommonModule, + GlobalConfigModule, + LoggerModule, + PlatformConfig, + ContextInterceptorModule, + CacheModule.register({ host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }) + ], + providers: [NatsService, JetStreamConsumer, TestSubsciber, Logger, PendingAckStore], exports: [NatsService] }) export class NatsModule implements OnModuleInit { @@ -21,6 +39,6 @@ export class NatsModule implements OnModuleInit { const jsm = this.nats.jetstreamManager(); await ensureStream(jsm); - await ensureConsumer(jsm); + // await ensureConsumer(jsm); } } diff --git a/apps/notification/src/nats/nats.service.ts b/apps/notification/src/nats/nats.service.ts index bb1a2009e..e2a1527b0 100644 --- a/apps/notification/src/nats/nats.service.ts +++ b/apps/notification/src/nats/nats.service.ts @@ -7,7 +7,8 @@ export class NatsService implements OnModuleDestroy { private js!: JetStreamClient; private jsm!: JetStreamManager; private connected = false; - private readonly logger = new Logger('NatsService'); + + constructor(private readonly logger: Logger) {} async connect(): Promise { if (this.connected) { diff --git a/apps/notification/src/nats/pendingAckStore.ts b/apps/notification/src/nats/pendingAckStore.ts new file mode 100644 index 000000000..3179a07a9 --- /dev/null +++ b/apps/notification/src/nats/pendingAckStore.ts @@ -0,0 +1,38 @@ +import { Injectable } from '@nestjs/common'; +import { JsMsg } from 'nats'; + +type PendingKey = string; + +interface PendingMessage { + msg: JsMsg; + receivedAt: number; +} + +@Injectable() +export class PendingAckStore { + private readonly store = new Map(); + + makeKey(stream: string, consumer: string, streamSequence: number): PendingKey { + return `${stream}:${consumer}:${streamSequence}`; + } + + save(stream: string, consumer: string, msg: JsMsg): PendingKey { + const info = msg.info!; + const key = this.makeKey(stream, consumer, info.streamSequence); + + this.store.set(key, { + msg, + receivedAt: Date.now() + }); + + return key; + } + + get(key: PendingKey): JsMsg | undefined { + return this.store.get(key)?.msg; + } + + delete(key: PendingKey): void { + this.store.delete(key); + } +} From d57ad52e3ebc7f4702b52d2d721b3d570e2aa35b Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Fri, 30 Jan 2026 16:52:29 +0530 Subject: [PATCH 05/13] feat: changes for consumer and sending notification to user Signed-off-by: Rinkal Bhojani --- apps/notification/package.json | 9 + .../src/nats/jetstream.consumer.ts | 113 +++- apps/notification/src/nats/jetstream.setup.ts | 122 +++- apps/notification/src/nats/nats-subscriber.ts | 52 +- apps/notification/src/nats/nats.module.ts | 12 +- pnpm-lock.yaml | 526 +++++++++++++++++- 6 files changed, 810 insertions(+), 24 deletions(-) create mode 100644 apps/notification/package.json diff --git a/apps/notification/package.json b/apps/notification/package.json new file mode 100644 index 000000000..ec5ff9b37 --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index 4952c68c0..80498c57b 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -1,18 +1,45 @@ import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { Consumer } from 'nats'; import { NatsService } from './nats.service'; +import { + DID_STREAM, + ensureConsumer, + ensureDidStream, + publishToJetStream, + PULL_CONSUMER, + 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'; @Injectable() export class JetStreamConsumer implements OnApplicationBootstrap { constructor( private readonly nats: NatsService, - private readonly logger: Logger + private readonly logger: Logger, + private readonly pendingAckStore: PendingAckStore, + private readonly holderNotificationRepository: HolderNotificationRepository ) {} async onApplicationBootstrap(): Promise { - // const js = this.nats.jetstream(); // ✅ now safe + if (!admin.apps.length) { + admin.initializeApp({ + credential: admin.credential.cert({ + projectId: process.env.PROJECT_ID, + clientEmail: process.env.CLIENT_EMAIL, + privateKey: process.env.PRIVATE_KEY.replace(/\\n/g, '\n') + }) + // credential: admin.credential.cert('/home/rinkal-bhojani/Rinkal/platform/dyuat-80c22-firebase-adminsdk-41x7t-3e8d2469ee.json') + }); + } + + const js = this.nats.jetstream(); // ✅ now safe - // const consumer: Consumer = await js.consumers.get(STREAM, CONSUMER); + const jsm = this.nats.jetstreamManager(); + await ensureConsumer(jsm); + const consumer: Consumer = await js.consumers.get(STREAM, PULL_CONSUMER); // const consumer: Consumer = await js.consumers.get( // STREAM, // { @@ -23,9 +50,9 @@ export class JetStreamConsumer implements OnApplicationBootstrap { this.logger.log('[NATS] JetStream consumer started'); - // this.consume(consumer).catch((err) => { - // this.logger.error('[NATS] Consumer crashed', err); - // }); + this.consume(consumer).catch((err) => { + this.logger.error('[NATS] Consumer crashed', err); + }); } private async consume(consumer: Consumer): Promise { @@ -35,6 +62,7 @@ export class JetStreamConsumer implements OnApplicationBootstrap { 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, @@ -44,13 +72,78 @@ export class JetStreamConsumer implements OnApplicationBootstrap { }); const payload = msg.json(); - this.logger.log('[NATS] Message received', payload); + const notificationDetail = await this.holderNotificationRepository.getHolderNotificationBySessionId(sessionId); + this.logger.log(`[NATS] Message received ${JSON.stringify(payload)}`); + this.logger.log(`[NATS] Processing message, ${JSON.stringify({ deliveryCount: msg.info.deliveryCount })}`); + this.logger.log(`[NATS] Consdition....', ${JSON.stringify({ IsFCMSatisfied: 3 < msg.info.deliveryCount })}`); + if (3 < msg.info.deliveryCount) { + //------------ Moving message to DID stream ---------------// + 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)}`); + + //------------- Sending Push Notification via FCM ----------------// + let notificationTitle = ''; + if ('presentation.ack' === event) { + notificationTitle = `Your data delivered to ${orgCode}`; + } else if ('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 the 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); + }); + msg.ack(); + } else { + // ------------- Publishing Messages via NATS ----------------// + + const ackKey = this.pendingAckStore.save('notify', consumerName, msg); + + this.logger.log('[NATS] Notification detail fetched', notificationDetail); + if (!notificationDetail) { + this.logger.error('[NATS] No notification detail found for session ID:', sessionId); + msg.nak(); + return; + } + await this.nats.publish(`${notificationDetail.holderDid}`, { + //`${notificationDetail.holderDid}`, { + payload, + ackKey, // opaque token user must return + subject: msg.subject, + event: `${domain}.${event}` + }); + // await this.nats.publish('did-key', payload); // ACK reply to APP - await this.nats.publish('did-key', payload); // ACK reply to APP - this.logger.log('[NATS] Message published to did:key:z6Mkm4VPWQRRSJSgoPVQ2oB8EfRqDGDQJBudpqppovJYeLpu'); + this.logger.log(`[NATS] Message published to ${notificationDetail.holderDid} for session ${sessionId}`); + } // business logic - msg.ack(); + // msg.ack(); } catch (err) { this.logger.error('[NATS] Processing failed', err); msg.nak(); diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index b1060e7ee..7da124637 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -1,11 +1,23 @@ /* eslint-disable camelcase */ import { Logger } from '@nestjs/common'; -import { AckPolicy, DeliverPolicy, JetStreamManager, ReplayPolicy, RetentionPolicy, StorageType, nanos } from 'nats'; +import { + AckPolicy, + DeliverPolicy, + JetStreamClient, + JetStreamManager, + ReplayPolicy, + RetentionPolicy, + StorageType, + nanos +} from 'nats'; +import assert = require('node:assert'); export const SOURCE_STREAM = 'notify'; export const STREAM = 'aggregate'; +export const DID_STREAM = 'did-notify'; export const SUBJECTS = ['aggregate.>']; export const CONSUMER = 'hub-consumer'; +export const PULL_CONSUMER = 'hub-pull-consumer'; const logger = new Logger('JetStreamSetup'); /** @@ -31,6 +43,22 @@ export async function ensureStream(jsm: JetStreamManager): Promise { } } +export async function ensureDidStream(jsm: JetStreamManager): Promise { + try { + // await jsm.streams.delete(DID_STREAM); + await jsm.streams.info(DID_STREAM); + logger.log(`[NATS] Stream '${DID_STREAM}' already exists`); + } catch { + logger.log(`[NATS] Creating stream '${DID_STREAM}'`); + await jsm.streams.add({ + name: DID_STREAM, + subjects: [`${DID_STREAM}.>`], + retention: RetentionPolicy.Workqueue, + storage: StorageType.File + }); + } +} + /** * Ensures the JetStream consumer exists */ @@ -41,17 +69,17 @@ export async function ensureConsumer(jsm: JetStreamManager): Promise { // // eslint-disable-next-line camelcase // // filter_subject: '*.presentation.test.blr' // }); - await jsm.consumers.info(STREAM, CONSUMER); - logger.log(`[NATS] Consumer '${CONSUMER}' already exists`); + await jsm.consumers.info(STREAM, PULL_CONSUMER); + logger.log(`[NATS] Consumer '${PULL_CONSUMER}' already exists`); } catch { logger.log(`[NATS] Creating consumer '${CONSUMER}'`); await jsm.consumers.add(STREAM, { - name: CONSUMER, - durable_name: CONSUMER, + name: PULL_CONSUMER, + durable_name: PULL_CONSUMER, ack_policy: AckPolicy.Explicit, deliver_policy: DeliverPolicy.All, - ack_wait: nanos(30_000), - max_deliver: 5 + ack_wait: nanos(10_000), + max_deliver: 4 // ,filter_subjects: ['hubStream.presentation.test.*'] }); } @@ -61,6 +89,7 @@ export async function ensureSessionConsumer(jsm: JetStreamManager, sessionId: st const consumerName = `notify-session-${sessionId}`; try { + // await jsm.consumers.delete(STREAM, consumerName); await jsm.consumers.info(STREAM, consumerName); logger.log(`[NATS] Consumer '${consumerName}' already exists`); return consumerName; @@ -76,10 +105,87 @@ export async function ensureSessionConsumer(jsm: JetStreamManager, sessionId: st deliver_policy: DeliverPolicy.All, filter_subject: `notify.*.*.*.${sessionId}`, replay_policy: ReplayPolicy.Instant, - max_ack_pending: 1000, + // max_ack_pending: 1000, + inactive_threshold: 30_000 + }); + + return consumerName; + } +} + +export async function ensureSessionConsumerWithDidStream(jsm: JetStreamManager, sessionId: string): Promise { + const consumerName = `notify-session-${sessionId}`; + + try { + await jsm.consumers.info(DID_STREAM, consumerName); + logger.log(`[NATS] Consumer '${consumerName}' already exists`); + return consumerName; + } catch (error) { + logger.error( + `Error creatingg consumer info for '${consumerName}': ${error instanceof Error ? error.message : 'Unknown error'}` + ); + logger.log(`[NATS] Creating consumer '${consumerName}'`); + await jsm.consumers.add(DID_STREAM, { + name: consumerName, + // durable_name: consumerName, + ack_policy: AckPolicy.Explicit, + deliver_policy: DeliverPolicy.All, + filter_subject: `notify.*.*.*.${sessionId}`, + replay_policy: ReplayPolicy.Instant, + // max_ack_pending: 1000, inactive_threshold: 30_000 }); return consumerName; } } + +export function assertDefined(v: T, msg = 'Value must be defined'): asserts v is NonNullable { + assert(null !== v && v !== undefined, msg); +} + +/** + * Publish to JetStream with guaranteed delivery + * Returns acknowledgment from JetStream with stream name and sequence number + */ +export async function publishToJetStream( + subject: string, + payload: unknown, + jsc: JetStreamClient + // options?: JetStreamPublishOptions, +): Promise<{ stream: string; seq: number }> { + assertDefined(subject, 'Subject is required for publishing'); + assertDefined(payload, 'payload is required for publishing'); + + try { + // const js = await getJetStreamClient(); + logger.log(`------------Publishing to JetStream------------ ${subject} with payload: ${JSON.stringify(payload)}`); + const ack = await jsc.publish( + `${subject}`, + // sc.encode(JSON.stringify(payload)) + JSON.stringify(payload) + ); + + logger.log( + `[NATS] Message published to JetStream, ${JSON.stringify({ + subject, + stream: ack.stream, + sequence: ack.seq, + dataSize: payload.length, + duplicate: ack.duplicate || false + })}` + ); + + return { + stream: ack.stream, + seq: ack.seq + }; + } catch (error) { + logger.error(`------------${JSON.stringify(error)}`); + logger.error('[NATS] JetStream publish failed', { + subject, + error: error instanceof Error ? error.message : 'Unknown error' + }); + throw error; + } +} diff --git a/apps/notification/src/nats/nats-subscriber.ts b/apps/notification/src/nats/nats-subscriber.ts index 7d290e4ec..9071abd78 100644 --- a/apps/notification/src/nats/nats-subscriber.ts +++ b/apps/notification/src/nats/nats-subscriber.ts @@ -1,7 +1,7 @@ import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { NatsService } from './nats.service'; import { INatsUserRequestData } from './interfaces'; -import { ensureSessionConsumer, STREAM } from './jetstream.setup'; +import { DID_STREAM, ensureSessionConsumer, ensureSessionConsumerWithDidStream, STREAM } from './jetstream.setup'; import { JetStreamClient } from 'nats'; import { PendingAckStore } from './pendingAckStore'; @@ -124,4 +124,54 @@ export class TestSubsciber implements OnApplicationBootstrap { this.logger.error(`[NATS] Failed to ACK message ${JSON.stringify(err)}`); } } + + // ------ For did-notify stream testing ------ // + + private async activateUserConsumersWithDidNotify(data: INatsUserRequestData): Promise { + const { did, sessions } = data; + + const js = this.nats.jetstream(); + const jsm = this.nats.jetstreamManager(); + this.logger.log(`[NATS] Started activating consumer for user DID: ${did} with sessions count: ${sessions.length}`); + + // Create all consumers in parallel first (dependency) + await Promise.all(sessions.map((session) => ensureSessionConsumerWithDidStream(jsm, session.sessionId))); + + // After all consumers are created, consume messages from all sessions in parallel + await Promise.all(sessions.map((session) => this.consumeSessionMessagesWithDidNotify(js, session.sessionId, did))); + } + + private async consumeSessionMessagesWithDidNotify( + js: JetStreamClient, + sessionId: string, + did: string + ): Promise { + const consumerName = `notify-session-${sessionId}`; + this.logger.log(`[NATS] Getting consumer ${consumerName} to fetch messages for session ${sessionId}`); + const consumer = await js.consumers.get(DID_STREAM, consumerName); + + const iter = await consumer.fetch({ + // eslint-disable-next-line camelcase + max_messages: 10 + }); + this.logger.log(`[NATS] Started fetching messages for consumer ${consumerName}`); + for await (const msg of iter) { + const payload = msg.json(); + // eslint-disable-next-line @typescript-eslint/no-unused-vars + const [_, domain, event, orgCode, sessionId] = msg.subject.split('.'); + + const ackKey = this.pendingAckStore.save('notify', consumerName, msg); + + await this.nats.publish(`${did}`, { + payload, + ackKey, // opaque token user must return + subject: msg.subject, + event: `${domain}.${event}` + }); + this.logger.log(`[NATS] Published message to user DID ${did} for session ${sessionId}`); + } + this.logger.log(`[NATS] Finished processing messages for session ${sessionId}, deleting consumer ${consumerName}`); + await this.nats.jetstreamManager().consumers.delete(STREAM, consumerName); + this.logger.log(`[NATS] Deleted consumer ${consumerName} after processing messages for session ${sessionId}`); + } } diff --git a/apps/notification/src/nats/nats.module.ts b/apps/notification/src/nats/nats.module.ts index 69b7bca81..a3c04d74f 100644 --- a/apps/notification/src/nats/nats.module.ts +++ b/apps/notification/src/nats/nats.module.ts @@ -11,6 +11,8 @@ import { LoggerModule } from '@credebl/logger'; import { ContextInterceptorModule } from '@credebl/context'; import { CacheModule } from '@nestjs/cache-manager'; import { PendingAckStore } from './pendingAckStore'; +import { HolderNotificationRepository } from '../holder-notification.repository'; +import { PrismaService } from '@credebl/prisma-service'; @Module({ imports: [ @@ -23,7 +25,15 @@ import { PendingAckStore } from './pendingAckStore'; ContextInterceptorModule, CacheModule.register({ host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }) ], - providers: [NatsService, JetStreamConsumer, TestSubsciber, Logger, PendingAckStore], + providers: [ + NatsService, + JetStreamConsumer, + TestSubsciber, + Logger, + PendingAckStore, + HolderNotificationRepository, + PrismaService + ], exports: [NatsService] }) export class NatsModule implements OnModuleInit { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4e01c3197..c1143fdc3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -469,6 +469,12 @@ importers: specifier: ^5.1.6 version: 5.9.2 + apps/notification: + dependencies: + firebase-admin: + specifier: ^13.6.0 + version: 13.6.0 + libs/aws: dependencies: '@nestjs/common': @@ -908,10 +914,70 @@ packages: resolution: {integrity: sha512-d9zaMRSTIKDLhctzH12MtXvJKSSUhaHcjV+2Z+GK+EEY7XKpP5yR4x+N3TAcHTcu963nIr+TMcCb4DBCYX1z6Q==} engines: {node: ^12.22.0 || ^14.17.0 || >=16.0.0} + '@fastify/busboy@3.2.0': + resolution: {integrity: sha512-m9FVDXU3GT2ITSe0UaMA5rU3QkfC/UXtCU8y0gSN/GugTqtVldOBWIB5V6V3sbmenVZUIpU6f+mPEO2+m5iTaA==} + + '@firebase/app-check-interop-types@0.3.3': + resolution: {integrity: sha512-gAlxfPLT2j8bTI/qfe3ahl2I2YcBQ8cFIBdhAQA4I2f3TndcO+22YizyGYuttLHPQEpWkhmpFW60VCFEPg4g5A==} + + '@firebase/app-types@0.9.3': + resolution: {integrity: sha512-kRVpIl4vVGJ4baogMDINbyrIOtOxqhkZQg4jTq3l8Lw6WSk0xfpEYzezFu+Kl4ve4fbPl79dvwRtaFqAC/ucCw==} + + '@firebase/auth-interop-types@0.2.4': + resolution: {integrity: sha512-JPgcXKCuO+CWqGDnigBtvo09HeBs5u/Ktc2GaFj2m01hLarbxthLNm7Fk8iOP1aqAtXV+fnnGj7U28xmk7IwVA==} + + '@firebase/component@0.7.0': + resolution: {integrity: sha512-wR9En2A+WESUHexjmRHkqtaVH94WLNKt6rmeqZhSLBybg4Wyf0Umk04SZsS6sBq4102ZsDBFwoqMqJYj2IoDSg==} + engines: {node: '>=20.0.0'} + + '@firebase/database-compat@2.1.0': + resolution: {integrity: sha512-8nYc43RqxScsePVd1qe1xxvWNf0OBnbwHxmXJ7MHSuuTVYFO3eLyLW3PiCKJ9fHnmIz4p4LbieXwz+qtr9PZDg==} + engines: {node: '>=20.0.0'} + + '@firebase/database-types@1.0.16': + resolution: {integrity: sha512-xkQLQfU5De7+SPhEGAXFBnDryUWhhlFXelEg2YeZOQMCdoe7dL64DDAd77SQsR+6uoXIZY5MB4y/inCs4GTfcw==} + + '@firebase/database@1.1.0': + resolution: {integrity: sha512-gM6MJFae3pTyNLoc9VcJNuaUDej0ctdjn3cVtILo3D5lpp0dmUHHLFN/pUKe7ImyeB1KAvRlEYxvIHNF04Filg==} + engines: {node: '>=20.0.0'} + + '@firebase/logger@0.5.0': + resolution: {integrity: sha512-cGskaAvkrnh42b3BA3doDWeBmuHFO/Mx5A83rbRDYakPjO9bJtRL3dX7javzc2Rr/JHZf4HlterTW2lUkfeN4g==} + engines: {node: '>=20.0.0'} + + '@firebase/util@1.13.0': + resolution: {integrity: sha512-0AZUyYUfpMNcztR5l09izHwXkZpghLgCUaAGjtMwXnCg3bj4ml5VgiwqOMOxJ+Nw4qN/zJAaOQBcJ7KGkWStqQ==} + engines: {node: '>=20.0.0'} + + '@google-cloud/firestore@7.11.6': + resolution: {integrity: sha512-EW/O8ktzwLfyWBOsNuhRoMi8lrC3clHM5LVFhGvO1HCsLozCOOXRAlHrYBoE6HL42Sc8yYMuCb2XqcnJ4OOEpw==} + engines: {node: '>=14.0.0'} + + '@google-cloud/paginator@5.0.2': + resolution: {integrity: sha512-DJS3s0OVH4zFDB1PzjxAsHqJT6sKVbRwwML0ZBP9PbU7Yebtu/7SWMRzvO2J3nUi9pRNITCfu4LJeooM2w4pjg==} + engines: {node: '>=14.0.0'} + + '@google-cloud/projectify@4.0.0': + resolution: {integrity: sha512-MmaX6HeSvyPbWGwFq7mXdo0uQZLGBYCwziiLIGq5JVX+/bdI3SAq6bP98trV5eTWfLuvsMcIC1YJOF2vfteLFA==} + engines: {node: '>=14.0.0'} + + '@google-cloud/promisify@4.0.0': + resolution: {integrity: sha512-Orxzlfb9c67A15cq2JQEyVc7wEsmFBmHjZWZYQMUyJ1qivXyMwdyNOs9odi79hze+2zqdTtu1E19IM/FtqZ10g==} + engines: {node: '>=14'} + + '@google-cloud/storage@7.18.0': + resolution: {integrity: sha512-r3ZwDMiz4nwW6R922Z1pwpePxyRwE5GdevYX63hRmAQUkUQJcBH/79EnQPDv5cOv1mFBgevdNWQfi3tie3dHrQ==} + engines: {node: '>=14'} + '@grpc/grpc-js@1.14.0': resolution: {integrity: sha512-N8Jx6PaYzcTRNzirReJCtADVoq4z7+1KQ4E70jTg/koQiMoUSN1kbNjPOqpPbhMFhfU1/l7ixspPl8dNY+FoUg==} engines: {node: '>=12.10.0'} + '@grpc/proto-loader@0.7.15': + resolution: {integrity: sha512-tMXdRCfYVixjuFK+Hk0Q1s38gV9zDiDJfWL3h1rv4Qc39oILCu1TRTDt7+fGUI8K4G1Fj125Hx/ru3azECWTyQ==} + engines: {node: '>=6'} + hasBin: true + '@grpc/proto-loader@0.8.0': resolution: {integrity: sha512-rc1hOQtjIWGxcxpb9aHAfLpIctjEnsDehj0DAiVfBlmT84uvR0uUtN2hEi/ecvWVjXUGf5qPF4qEgiLOx1YIMQ==} engines: {node: '>=6'} @@ -1993,6 +2059,10 @@ packages: '@tokenizer/token@0.3.0': resolution: {integrity: sha512-OvjF+z51L3ov0OyAU0duzsYuvO01PH7x4t6DJx+guahgTnBHkhJdG7soQeTSFLWN3efnHyibZ4Z8l2EuWwJN3A==} + '@tootallnate/once@2.0.0': + resolution: {integrity: sha512-XCuKFP5PS55gnMVu3dty8KPatLqUoy/ZYzDzAGCQ8JNFCkLXzmI7vNHCR+XpbZaMWQK/vQubr7PkYq8g470J/A==} + engines: {node: '>= 10'} + '@tootallnate/quickjs-emscripten@0.23.0': resolution: {integrity: sha512-C5Mc6rdnsaJDjO3UpGW/CQTHtCKaYlScZTly4JIu97Jxo/odCiH0ITnDXSJPTOrEKk/ycSZ0AOgTmkDtkOsvIA==} @@ -2026,6 +2096,9 @@ packages: '@types/body-parser@1.19.6': resolution: {integrity: sha512-HLFeCYgz89uk22N5Qg3dvGvsv46B8GLvKKo1zKG4NybA8U2DiEO3w9lqGg29t/tfLRJpJ6iQxnVw4OnB7MoM9g==} + '@types/caseless@0.12.5': + resolution: {integrity: sha512-hWtVTC2q7hc7xZ/RLbxapMvDMgUnDvKvMOpKal4DrMyfGBUfB1oKaZlIRr6mJL+If3bAP6sV/QneGzF6tJjZDg==} + '@types/connect@3.4.38': resolution: {integrity: sha512-K6uROf1LD88uDQqJCktA4yzL1YYAK6NgfsI0v/mTgyPKWsX1CnJ0XPSDhViejru1GcRkLWb8RlzFYJRqGUbaug==} @@ -2086,6 +2159,9 @@ packages: '@types/jsonwebtoken@9.0.7': resolution: {integrity: sha512-ugo316mmTYBl2g81zDFnZ7cfxlut3o+/EQdaP7J8QN2kY6lJ22hmQYCK5EHcJHbrW+dkCGSCPgbG8JtYj6qSrg==} + '@types/long@4.0.2': + resolution: {integrity: sha512-MqTGEo5bj5t157U6fA/BiDynNkn0YknVdh48CMPkTSpFTVmvao5UQmm7uEF6xBEo7qIMAlY/JSleYaE6VOdpaA==} + '@types/luxon@3.7.1': resolution: {integrity: sha512-H3iskjFIAn5SlJU7OuxUmTEpebK6TKB8rxZShDslBMZJ5u9S//KM1sbdAisiSrqwLQncVjnpi2OK2J51h+4lsg==} @@ -2131,6 +2207,9 @@ packages: '@types/range-parser@1.2.7': resolution: {integrity: sha512-hKormJbkJqzQGhziax5PItDUTMAM9uE2XXQmM37dyd4hVM+5aVl7oVxMVUiVQn2oCQFN/LKCZdvSM0pFRqbSmQ==} + '@types/request@2.48.13': + resolution: {integrity: sha512-FGJ6udDNUCjd19pp0Q3iTiDkwhYup7J8hpMW9c4k53NrccQFFWKRho6hvtPPEhnXWKvukfwAlB6DbDz4yhH5Gg==} + '@types/retry@0.12.5': resolution: {integrity: sha512-3xSjTp3v03X/lSQLkczaN9UIEwJMoMCA1+Nb5HfbJEQWogdeQIyVtTvxPXDQjZ5zws8rFQfVfRdz03ARihPJgw==} @@ -2152,6 +2231,9 @@ packages: '@types/supertest@2.0.16': resolution: {integrity: sha512-6c2ogktZ06tr2ENoZivgm7YnprnhYE4ZoXGMY+oA7IuAf17M8FWvujXZGmxLv8y0PTyts4x5A+erSwVUFA8XSg==} + '@types/tough-cookie@4.0.5': + resolution: {integrity: sha512-/Ad8+nIOV7Rl++6f1BdKxFSMgmoqEoYbHRpPcx3JEfv8VRsQe9Z4mCXeJBzxs7mbHY/XOZZuXlRNfhpVPbs6ZA==} + '@types/triple-beam@1.3.5': resolution: {integrity: sha512-6WaYesThRMCl19iryMYP7/x2OVgCtbIVflDGFpWnb9irXI3UjYE4AzmYuiUKY1AJstGijoY+MgUszMgRxIYTYw==} @@ -2316,6 +2398,10 @@ packages: abbrev@1.1.1: resolution: {integrity: sha512-nne9/IiQ/hzIhY6pdDnbBtz7DjPTKrY00P/zvPSm5pOFkl6xuGrGnXn/VtTNNfNtAfZ9/1RtehkszU9qcTii0Q==} + abort-controller@3.0.0: + resolution: {integrity: sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg==} + engines: {node: '>=6.5'} + accepts@1.3.8: resolution: {integrity: sha512-PYAthTa2m2VKxuvSD3DPC/Gy+U+sOA1LAuT8mkmRuvw+NACSaeXEQ+NHcVF7rONl6qcaxV3Uuemwawk+7+SJLw==} engines: {node: '>= 0.6'} @@ -2504,6 +2590,10 @@ packages: resolution: {integrity: sha512-BNoCY6SXXPQ7gF2opIP4GBE+Xw7U+pHMYKuzjgCN3GwiaIR09UUeKfheyIry77QtrCBlC0KK0q5/TER/tYh3PQ==} engines: {node: '>= 0.4'} + arrify@2.0.1: + resolution: {integrity: sha512-3duEwti880xqi4eAMN8AyR4a0ByT90zoYdLlevfrvU43vb0YZwZVfxOgxWrLXXXpyugL0hNZc9G6BiB5B3nUug==} + engines: {node: '>=8'} + asap@2.0.6: resolution: {integrity: sha512-BSHWgDSAiKs50o2Re8ppvp3seVHXSRM44cdSsT9FfNEUUZLOGWVCsiWaRPWM1Znn+mqZ1OfVZ3z3DWEzSp7hRA==} @@ -2641,6 +2731,9 @@ packages: resolution: {integrity: sha512-QxD8cf2eVqJOOz63z6JIN9BzvVs/dlySa5HGSBH5xtR8dPteIRQnBxxKqkNTiT6jbDTF6jAfrd4oMcND9RGbQg==} engines: {node: '>=0.6'} + bignumber.js@9.3.1: + resolution: {integrity: sha512-Ko0uX15oIUS7wJ3Rb30Fs6SkVbLmPBAKdlm7q9+ak9bbIeFf0MwuBsQV6z7+X768/cHsfg+WlysDWJcmthjsjQ==} + binary-search@1.3.6: resolution: {integrity: sha512-nbE1WxOTTrUWIfsfZ4aHGYu5DOuNkbxGokjV6Z2kxfJK3uaAb8zNK1muzOeipoLHZjInT4Br88BHpzevc681xA==} @@ -3252,6 +3345,9 @@ packages: duplexer2@0.1.4: resolution: {integrity: sha512-asLFVfWWtJ90ZyOUHMqk7/S2w2guQKxUI2itj3d92ADHhxUSbCMGi1f1cBcJ7xM1To+pE/Khbwo1yuNbMEPKeA==} + duplexify@4.1.3: + resolution: {integrity: sha512-M3BmBhwJRZsSx38lZyhE53Csddgzl5R7xGJNk7CVddZD6CcmwMCH8J+7AprIrQKH7TonKxaCjcv27Qmf+sQ+oA==} + eastasianwidth@0.2.0: resolution: {integrity: sha512-I88TYZWc9XiYHRQ4/3c5rjjfgkjhLyW2luGIheGERbNQ6OY7yTybanSpDXZa8y7VUP9YmDcYa+eyq4ca7iLqWA==} @@ -3539,6 +3635,10 @@ packages: resolution: {integrity: sha512-aIL5Fx7mawVa300al2BnEE4iNvo1qETxLrPI/o05L7z6go7fCw1J6EQmbK4FmJ2AS7kgVF/KEZWufBfdClMcPg==} engines: {node: '>= 0.6'} + event-target-shim@5.0.1: + resolution: {integrity: sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ==} + engines: {node: '>=6'} + eventemitter3@5.0.1: resolution: {integrity: sha512-GWkBvjiSZK87ELrYOSESUYeVIc9mvLLf/nXalMOS5dYrgZq9o5OVkbZAVM06CVxYsCwH9BDZFPlQTlPA1j4ahA==} @@ -3597,6 +3697,10 @@ packages: resolution: {integrity: sha512-11Ndz7Nv+mvAC1j0ktTa7fAb0vLyGGX+rMHNBYQviQDGU0Hw7lhctJANqbPhu9nV9/izT/IntTgZ7Im/9LJs9g==} engines: {'0': node >=0.6.0} + farmhash-modern@1.1.0: + resolution: {integrity: sha512-6ypT4XfgqJk/F3Yuv4SX26I3doUjt0GTG4a+JgWxXQpxXzTBq8fPUeGHfcYMMDPHJHm3yPOSjaeBwBGAHWXCdA==} + engines: {node: '>=18.0.0'} + fast-deep-equal@3.1.3: resolution: {integrity: sha512-f3qQ9oQy9j2AhBe/H9VC91wLmKBCCU/gDOnKNAYG5hswO7BLKj09Hc5HYNz9cGI++xlpDCIgDaitVs03ATR84Q==} @@ -3632,6 +3736,10 @@ packages: fast-uri@3.1.0: resolution: {integrity: sha512-iPeeDKJSWf4IEOasVVrknXpaBV0IApz/gp7S2bb7Z4Lljbl2MGJRqInZiUrQwV16cpzw/D3S5j5Julj/gT52AA==} + fast-xml-parser@4.5.3: + resolution: {integrity: sha512-RKihhV+SHsIUGXObeVy9AXiBbFwkVk7Syp8XgwN5U3JV416+Gwp/GO9i0JYKmikykgz/UHRrrV4ROuZEo/T0ig==} + hasBin: true + fast-xml-parser@5.2.5: resolution: {integrity: sha512-pfX9uG9Ki0yekDHx2SiuRIyFdyAr1kMIMitPvb0YBo8SUfKvia7w7FIyd/l6av85pFYRhZscS75MwMnbvY+hcQ==} hasBin: true @@ -3639,6 +3747,10 @@ packages: fastq@1.19.1: resolution: {integrity: sha512-GwLTyxkCXjXbxqIhTsMI2Nui8huMPtnxg7krajPJAjnEG/iiOS7i+zCtWGZR9G0NBKbXKh6X9m9UIsYX/N6vvQ==} + faye-websocket@0.11.4: + resolution: {integrity: sha512-CzbClwlXAuiRQAlUyfqPgvPoNKTckTPGfwZV4ZdAhVcP2lh9KUxJg2b5GkE7XbjKQ3YJnQ9z6D9ntLAlB+tP8g==} + engines: {node: '>=0.8.0'} + fb-watchman@2.0.2: resolution: {integrity: sha512-p5161BqbuCaSnB8jIbzQHOlpgsPmK5rJVDfDKO91Axs5NC1uu3HRQm6wt9cd9/+GtQQIO53JdGXXoyDpTAsgYA==} @@ -3682,6 +3794,10 @@ packages: resolution: {integrity: sha512-78/PXT1wlLLDgTzDs7sjq9hzz0vXD+zn+7wypEe4fXQxCmdmqfGsEPQxmiCSQI3ajFV91bVSsvNtrJRiW6nGng==} engines: {node: '>=10'} + firebase-admin@13.6.0: + resolution: {integrity: sha512-GdPA/t0+Cq8p1JnjFRBmxRxAGvF/kl2yfdhALl38PrRp325YxyQ5aNaHui0XmaKcKiGRFIJ/EgBNWFoDP0onjw==} + engines: {node: '>=18'} + flat-cache@3.2.0: resolution: {integrity: sha512-CYcENa+FtcUKLmhhqyctpclsq7QF38pKjZHsGNiSQF5r4FtoKDWabFDl3hzaEQMvT1LHEysw5twgLvpYYb4vbw==} engines: {node: ^10.12.0 || >=12.0.0} @@ -3729,6 +3845,10 @@ packages: resolution: {integrity: sha512-1lLKB2Mu3aGP1Q/2eCOx0fNbRMe7XdwktwOruhfqqd0rIJWwN4Dh+E3hrPSlDCXnSR7UtZ1N38rVXm+6+MEhJQ==} engines: {node: '>= 0.12'} + form-data@2.5.5: + resolution: {integrity: sha512-jqdObeR2rxZZbPSGL+3VckHMYtu+f9//KXBsVny6JSX/pa38Fy+bGjuG8eW/H6USNQWhLi8Num++cU2yOCNz4A==} + engines: {node: '>= 0.12'} + form-data@4.0.4: resolution: {integrity: sha512-KrGhL9Q4zjj0kiUt5OO4Mr/A/jlI2jDYs5eHBpYHPcBEVSiipAvn2Ko2HnPe20rmcuuvMHNdZFp+4IlGTMF0Ow==} engines: {node: '>= 6'} @@ -3788,6 +3908,9 @@ packages: resolution: {integrity: sha512-e5iwyodOHhbMr/yNrc7fDYG4qlbIvI5gajyzPnb5TCwyhjApznQh1BMFou9b30SevY43gCJKXycoCBjMbsuW0Q==} engines: {node: '>= 0.4'} + functional-red-black-tree@1.0.1: + resolution: {integrity: sha512-dsKNQNdj6xA3T+QlADDA7mOSlX0qiMINjn0cgr+eGHGsbSHzTabcIogz2+p/iqP1Xs6EP/sS2SbqH+brGTbq0g==} + functions-have-names@1.2.3: resolution: {integrity: sha512-xckBUXyTIqT97tq2x2AMb+g163b5JFysYk0x4qxNFwbfQkmNZoiRHb6sPzI9/QV33WeuvVYBUIiD4NzNIyqaRQ==} @@ -3796,6 +3919,14 @@ packages: engines: {node: '>=10'} deprecated: This package is no longer supported. + gaxios@6.7.1: + resolution: {integrity: sha512-LDODD4TMYx7XXdpwxAVRAIAuB0bzv0s+ywFonY46k126qzQHT9ygyoa9tncmOiQmmDrik65UYsEkv3lbfqQ3yQ==} + engines: {node: '>=14'} + + gcp-metadata@6.1.1: + resolution: {integrity: sha512-a4tiq7E0/5fTjxPAaH4jpjkSv/uCaU2p5KC6HVGrvl0cDjA8iBZv4vv1gyzlmK0ZUKqwpOyQMKzZQe3lTit77A==} + engines: {node: '>=14'} + generate-password@1.7.1: resolution: {integrity: sha512-9bVYY+16m7W7GczRBDqXE+VVuCX+bWNrfYKC/2p2JkZukFb2sKxT6E3zZ3mJGz7GMe5iRK0A/WawSL3jQfJuNQ==} @@ -3882,6 +4013,18 @@ packages: resolution: {integrity: sha512-jhIXaOzy1sb8IyocaruWSn1TjmnBVs8Ayhcy83rmxNJ8q2uWKCAj3CnJY+KpGSXCueAPc0i05kVvVKtP1t9S3g==} engines: {node: '>=10'} + google-auth-library@9.15.1: + resolution: {integrity: sha512-Jb6Z0+nvECVz+2lzSMt9u98UsoakXxA2HGHMCxh+so3n90XgYWkq5dur19JAJV7ONiJY22yBTyJB1TSkvPq9Ng==} + engines: {node: '>=14'} + + google-gax@4.6.1: + resolution: {integrity: sha512-V6eky/xz2mcKfAd1Ioxyd6nmA61gao3n01C+YeuIwu3vzM9EDR6wcVzMSIbLMDXWeoi9SHYctXuKYC5uJUT3eQ==} + engines: {node: '>=14'} + + google-logging-utils@0.0.2: + resolution: {integrity: sha512-NEgUnEcBiP5HrPzufUkBzJOD/Sxsco3rLNo1F1TNf7ieU8ryUzBhqba8r756CjLX7rn3fHl6iLEwPYuqpoKgQQ==} + engines: {node: '>=14'} + gopd@1.2.0: resolution: {integrity: sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==} engines: {node: '>= 0.4'} @@ -3892,6 +4035,10 @@ packages: graphemer@1.4.0: resolution: {integrity: sha512-EtKwoO6kxCL9WO5xipiHTZlSzBm7WLT627TqC/uVRd0HKmq8NXyebnNYxDoBi7wt8eTWrUrKXCOVaFq9x1kgag==} + gtoken@7.1.0: + resolution: {integrity: sha512-pCcEwRi+TKpMlxAQObHDQ56KawURgyAf6jtIY046fJ5tIv3zDe/LEIubckAO8fj6JnAxLdmWkUfNyulQ2iKdEw==} + engines: {node: '>=14.0.0'} + handlebars@4.7.8: resolution: {integrity: sha512-vafaFqs8MZkRrSX7sFVUdo3ap/eNiLnb4IakshzvP56X5Nr1iGKAIqdX6tMlm6HcNRIkr6AxO5jFEoJzzpT8aQ==} engines: {node: '>=0.4.7'} @@ -3951,6 +4098,9 @@ packages: hpagent@0.1.2: resolution: {integrity: sha512-ePqFXHtSQWAFXYmj+JtOTHr84iNrII4/QRlAAPPE+zqnKy4xJo7Ie1Y4kC7AdB+LxLxSTTzBMASsEcy0q8YyvQ==} + html-entities@2.6.0: + resolution: {integrity: sha512-kig+rMn/QOVRvr7c86gQ8lWXq+Hkv6CbAH1hLu+RG338StTpE8Z0b44SDVaqVu7HGKf27frdmUYEs9hTUX/cLQ==} + html-escaper@2.0.2: resolution: {integrity: sha512-H2iMtd0I4Mt5eYiapRdIDjp+XzelXQ0tFE4JS7YFwFevXXMmOp9myNrUvCg0D6ws8iqkRPBfKHgbwig1SmlLfg==} @@ -3970,6 +4120,13 @@ packages: http-headers@3.0.2: resolution: {integrity: sha512-87E1I+2Wg4dxxz4rcxElo3dxO/w1ZtgL1yA0Sb6vH3qU16vRKq1NjWQv9SCY3ly2OQROcoxHZOUpmelS+k6wOw==} + http-parser-js@0.5.10: + resolution: {integrity: sha512-Pysuw9XpUq5dVc/2SMHpuTY01RFl8fttgcyunjL7eEMhGM3cI4eOmiCycJDVCo/7O7ClfQD3SaI6ftDzqOXYMA==} + + http-proxy-agent@5.0.0: + resolution: {integrity: sha512-n2hY8YdoRE1i7r6M0w9DIw5GgZN0G25P8zLCRQ8rjXtTU3vsNFBI/vWK/UIeE6g5MUUz6avwAPXmL6Fy9D/90w==} + engines: {node: '>= 6'} + http-proxy-agent@7.0.2: resolution: {integrity: sha512-T1gkAiYYDWYx3V5Bmyu7HcfcvL7mUrTWiM6yOfa3PIphViJ/gFPbvidQ+veqSOHci/PxBcDabeUNCzpOODJZig==} engines: {node: '>= 14'} @@ -4453,6 +4610,9 @@ packages: engines: {node: '>=6'} hasBin: true + json-bigint@1.0.0: + resolution: {integrity: sha512-SiPv/8VpZuWbvLSMtTDU8hEfrZWg/mH/nV/b4o0CYbSxu1UIQPLdwKOCIyLQX+VIPO5vrLX3i8qtqFyhdPSUSQ==} + json-buffer@3.0.1: resolution: {integrity: sha512-4bV5BfR2mqfQTJm+V5tPPdf+ZpuhiIvTuAB5g8kcrXOZpTT/QwwVRWBywX1ozr6lEuPdbHxwaJlm9G6mI2sfSQ==} @@ -4800,6 +4960,11 @@ packages: engines: {node: '>=4.0.0'} hasBin: true + mime@3.0.0: + resolution: {integrity: sha512-jSCU7/VB1loIWBZe14aEYHU/+1UMEHoaO7qxCOVJOw9GgH72VAWppxNcjU+x9a2k3GSIBXNKxXQFqRvvZ7vr3A==} + engines: {node: '>=10.0.0'} + hasBin: true + mimic-fn@2.1.0: resolution: {integrity: sha512-OqbOk5oEQeAZ8WXWydlu9HJjz9WVdEIvamMCcXmuqUYjTknH/sqsWvhQ3vgwKFRR1HpjvNBKQ37nbJgYzGqGcg==} engines: {node: '>=6'} @@ -4983,6 +5148,10 @@ packages: encoding: optional: true + node-forge@1.3.3: + resolution: {integrity: sha512-rLvcdSyRCyouf6jcOIPe/BgwG/d7hKjzMKOas33/pHEr6gbq18IK9zV7DiPvzsz0oBJPme6qr6H6kGZuI9/DZg==} + engines: {node: '>= 6.13.0'} + node-gyp-build-optional-packages@5.2.2: resolution: {integrity: sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==} hasBin: true @@ -5395,6 +5564,10 @@ packages: resolution: {integrity: sha512-NxNv/kLguCA7p3jE8oL2aEBsrJWgAakBpgmgK6lpPWV+WuOmY6r2/zbAVnP+T8bQlA0nzHXSJSJW0Hq7ylaD2Q==} engines: {node: '>= 6'} + proto3-json-serializer@2.0.2: + resolution: {integrity: sha512-SAzp/O4Yh02jGdRc+uIrGoe87dkN/XtwxfZ4ZyafJHymd79ozp5VG5nyZ7ygqPM5+cpLDjjGnYFUkngonyDPOQ==} + engines: {node: '>=14.0.0'} + protobufjs@7.5.4: resolution: {integrity: sha512-CvexbZtbov6jW2eXAvLukXjXUW1TzFaivC46BpWc/3BpcCysb5Vffu+B3XHMm8lVEuy2Mm4XGex8hBSg1yapPg==} engines: {node: '>=12.0.0'} @@ -5446,12 +5619,12 @@ packages: puppeteer@21.0.1: resolution: {integrity: sha512-KTjmSdPZ6bMkq3EbAzAUhcB3gMDXvdwd6912rxG9hNtjwRJzHSA568vh6vIbO2WQeNmozRdt1LtiUMLSWfeMrg==} engines: {node: '>=16.3.0'} - deprecated: < 24.10.2 is no longer supported + deprecated: < 24.15.0 is no longer supported puppeteer@21.11.0: resolution: {integrity: sha512-9jTHuYe22TD3sNxy0nEIzC7ZrlRnDgeX3xPkbS7PnbdwYjl2o/z/YuCrRBwezdKpbTDTJ4VqIggzNyeRcKq3cg==} engines: {node: '>=16.13.2'} - deprecated: < 24.10.2 is no longer supported + deprecated: < 24.15.0 is no longer supported hasBin: true pure-rand@6.1.0: @@ -5624,6 +5797,10 @@ packages: restructure@2.0.1: resolution: {integrity: sha512-e0dOpjm5DseomnXx2M5lpdZ5zoHqF1+bqdMJUohoYVVQa7cBdnk7fdmeI6byNWP/kiME72EeTiSypTCVnpLiDg==} + retry-request@7.0.2: + resolution: {integrity: sha512-dUOvLMJ0/JJYEn8NrpOaGNE7X3vpI5XlZS/u0ANjqtcZVKnIxP7IgCFwrKTxENw29emmwug53awKtaMm4i9g5w==} + engines: {node: '>=14'} + retry@0.13.1: resolution: {integrity: sha512-XQBQ3I8W1Cge0Seh+6gjj03LbmRFWuoszgK9ooCpwYIrhhoO80pfq4cUkU5DkknwfOfFteRwlZ56PYOGYyFWdg==} engines: {node: '>= 4'} @@ -5910,6 +6087,12 @@ packages: stream-chopper@3.0.1: resolution: {integrity: sha512-f7h+ly8baAE26iIjcp3VbnBkbIRGtrvV0X0xxFM/d7fwLTYnLzDPTXRKNxa2HZzohOrc96NTrR+FaV3mzOelNA==} + stream-events@1.0.5: + resolution: {integrity: sha512-E1GUzBSgvct8Jsb3v2X15pjzN1tYebtbLaMg+eBOUOAxgbLoSbT2NS91ckc5lJD1KfLjId+jXJRgo0qnV5Nerg==} + + stream-shift@1.0.3: + resolution: {integrity: sha512-76ORR0DO1o1hlKwTbi/DM3EXWGf3ZJYO8cXX5RJwnul2DEg2oyoZyjLNoQM8WsvZiFKCRfC1O0J7iCvie3RZmQ==} + streamsearch@1.1.0: resolution: {integrity: sha512-Mcc5wHehp9aXz1ax6bZUyY5afg9u2rv5cqQI3mRrYkGC8rW2hM02jWuwjtL++LS5qinSyhj2QfLyNsuc+VsExg==} engines: {node: '>=10.0.0'} @@ -5979,6 +6162,9 @@ packages: resolution: {integrity: sha512-6fPc+R4ihwqP6N/aIv2f1gMH8lOVtWQHoqC4yK6oSDVVocumAsfCqjkXnqiYMhmMwS/mEHLp7Vehlt3ql6lEig==} engines: {node: '>=8'} + strnum@1.1.2: + resolution: {integrity: sha512-vrN+B7DBIoTTZjnPNewwhx6cBA/H+IS7rfW68n7XxC1y7uoiGQBxaKzqucGUgavX15dJgiGztLJ8vxuEzwqBdA==} + strnum@2.1.1: resolution: {integrity: sha512-7ZvoFTiCnGxBtDqJ//Cu6fWtZtc7Y3x+QOirG15wztbdngGSkht27o2pyGWrVy0b4WAy3jbKmnoK6g5VlVNUUw==} @@ -5986,6 +6172,9 @@ packages: resolution: {integrity: sha512-KIy5nylvC5le1OdaaoCJ07L+8iQzJHGH6pWDuzS+d07Cu7n1MZ2x26P8ZKIWfbK02+XIL8Mp4RkWeqdUCrDMfg==} engines: {node: '>=18'} + stubs@3.0.0: + resolution: {integrity: sha512-PdHt7hHUJKxvTCgbKX9C1V/ftOcjJQgz8BZwNfV5c4B6dcGqlpelTbJ999jBGZ2jYiPAwcX5dP6oBwVlBlUbxw==} + superagent@7.1.6: resolution: {integrity: sha512-gZkVCQR1gy/oUXr+kxJMLDjla434KmSOKbx5iGD30Ql+AkJQ/YlPKECJy2nhqOsHLjGHzoDTXNSjhnvWhzKk7g==} engines: {node: '>=6.4.0 <13 || >=14'} @@ -6045,6 +6234,11 @@ packages: tar@6.2.1: resolution: {integrity: sha512-DZ4yORTwrbTj/7MZYq2w+/ZFdI6OZ/f9SFHR+71gIVUZhOQPHzVCLpvRnPgyaMpfWxxk/4ONva3GQSyNIKRv6A==} engines: {node: '>=10'} + deprecated: Old versions of tar are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exhorbitant rates) by contacting i@izs.me + + teeny-request@9.0.0: + resolution: {integrity: sha512-resvxdc6Mgb7YEThw6G6bExlXKkv6+YbuzGg9xuXxSgxJF7Ozs+o8Y9+2R3sArdWdW8nOokoQb1yrpFB0pQK2g==} + engines: {node: '>=14'} telejson@7.2.0: resolution: {integrity: sha512-1QTEcJkJEhc8OnStBx/ILRu5J2p0GjvWsBx56bmZRqnrkdBMUe+nX92jxV+p3dB4CP6PZCdJMQJwCggkNBMzkQ==} @@ -6508,6 +6702,14 @@ packages: webpack-cli: optional: true + websocket-driver@0.7.4: + resolution: {integrity: sha512-b17KeDIQVjvb0ssuSDF2cYXSg2iztliJ4B9WdsuB6J952qCPKmnVq4DyW5motImXHDC1cBT/1UezrJVsKw5zjg==} + engines: {node: '>=0.8.0'} + + websocket-extensions@0.1.4: + resolution: {integrity: sha512-OqedPIGOfsDlo31UNwYbCFMSaO9m9G/0faIHj5/dZFDMFqPTcx6UwqyOy3COEaEOg/9VsGIpdqn62W5KhoKSpg==} + engines: {node: '>=0.8.0'} + whatwg-url@5.0.0: resolution: {integrity: sha512-saE57nupxk6v3HY35+jzBwYa0rKSy0XR8JSxZPwgLr7ys0IBzhGviA1/TUGJLmSVqs8pb9AnvICXEuOHLprYTw==} @@ -7355,11 +7557,110 @@ snapshots: '@eslint/js@8.57.1': {} + '@fastify/busboy@3.2.0': {} + + '@firebase/app-check-interop-types@0.3.3': {} + + '@firebase/app-types@0.9.3': {} + + '@firebase/auth-interop-types@0.2.4': {} + + '@firebase/component@0.7.0': + dependencies: + '@firebase/util': 1.13.0 + tslib: 2.8.1 + + '@firebase/database-compat@2.1.0': + dependencies: + '@firebase/component': 0.7.0 + '@firebase/database': 1.1.0 + '@firebase/database-types': 1.0.16 + '@firebase/logger': 0.5.0 + '@firebase/util': 1.13.0 + tslib: 2.8.1 + + '@firebase/database-types@1.0.16': + dependencies: + '@firebase/app-types': 0.9.3 + '@firebase/util': 1.13.0 + + '@firebase/database@1.1.0': + dependencies: + '@firebase/app-check-interop-types': 0.3.3 + '@firebase/auth-interop-types': 0.2.4 + '@firebase/component': 0.7.0 + '@firebase/logger': 0.5.0 + '@firebase/util': 1.13.0 + faye-websocket: 0.11.4 + tslib: 2.8.1 + + '@firebase/logger@0.5.0': + dependencies: + tslib: 2.8.1 + + '@firebase/util@1.13.0': + dependencies: + tslib: 2.8.1 + + '@google-cloud/firestore@7.11.6': + dependencies: + '@opentelemetry/api': 1.9.0 + fast-deep-equal: 3.1.3 + functional-red-black-tree: 1.0.1 + google-gax: 4.6.1 + protobufjs: 7.5.4 + transitivePeerDependencies: + - encoding + - supports-color + optional: true + + '@google-cloud/paginator@5.0.2': + dependencies: + arrify: 2.0.1 + extend: 3.0.2 + optional: true + + '@google-cloud/projectify@4.0.0': + optional: true + + '@google-cloud/promisify@4.0.0': + optional: true + + '@google-cloud/storage@7.18.0': + dependencies: + '@google-cloud/paginator': 5.0.2 + '@google-cloud/projectify': 4.0.0 + '@google-cloud/promisify': 4.0.0 + abort-controller: 3.0.0 + async-retry: 1.3.3 + duplexify: 4.1.3 + fast-xml-parser: 4.5.3 + gaxios: 6.7.1 + google-auth-library: 9.15.1 + html-entities: 2.6.0 + mime: 3.0.0 + p-limit: 3.1.0 + retry-request: 7.0.2 + teeny-request: 9.0.0 + uuid: 8.3.2 + transitivePeerDependencies: + - encoding + - supports-color + optional: true + '@grpc/grpc-js@1.14.0': dependencies: '@grpc/proto-loader': 0.8.0 '@js-sdsl/ordered-map': 4.4.2 + '@grpc/proto-loader@0.7.15': + dependencies: + lodash.camelcase: 4.3.0 + long: 5.3.2 + protobufjs: 7.5.4 + yargs: 17.7.2 + optional: true + '@grpc/proto-loader@0.8.0': dependencies: lodash.camelcase: 4.3.0 @@ -8761,6 +9062,9 @@ snapshots: '@tokenizer/token@0.3.0': {} + '@tootallnate/once@2.0.0': + optional: true + '@tootallnate/quickjs-emscripten@0.23.0': {} '@tsconfig/node10@1.0.11': {} @@ -8801,6 +9105,9 @@ snapshots: '@types/connect': 3.4.38 '@types/node': 20.19.17 + '@types/caseless@0.12.5': + optional: true + '@types/connect@3.4.38': dependencies: '@types/node': 20.19.17 @@ -8877,6 +9184,9 @@ snapshots: dependencies: '@types/node': 20.19.17 + '@types/long@4.0.2': + optional: true + '@types/luxon@3.7.1': {} '@types/methods@1.1.4': {} @@ -8928,6 +9238,14 @@ snapshots: '@types/range-parser@1.2.7': {} + '@types/request@2.48.13': + dependencies: + '@types/caseless': 0.12.5 + '@types/node': 20.19.17 + '@types/tough-cookie': 4.0.5 + form-data: 2.5.5 + optional: true + '@types/retry@0.12.5': {} '@types/semver@7.7.1': {} @@ -8956,6 +9274,9 @@ snapshots: dependencies: '@types/superagent': 8.1.9 + '@types/tough-cookie@4.0.5': + optional: true + '@types/triple-beam@1.3.5': {} '@types/validator@13.15.3': {} @@ -9183,6 +9504,11 @@ snapshots: abbrev@1.1.1: {} + abort-controller@3.0.0: + dependencies: + event-target-shim: 5.0.1 + optional: true + accepts@1.3.8: dependencies: mime-types: 2.1.35 @@ -9369,6 +9695,9 @@ snapshots: get-intrinsic: 1.3.0 is-array-buffer: 3.0.5 + arrify@2.0.1: + optional: true + asap@2.0.6: {} asn1.js@5.4.1: @@ -9548,6 +9877,8 @@ snapshots: big-integer@1.6.52: {} + bignumber.js@9.3.1: {} + binary-search@1.3.6: optional: true @@ -10172,6 +10503,14 @@ snapshots: dependencies: readable-stream: 2.3.8 + duplexify@4.1.3: + dependencies: + end-of-stream: 1.4.5 + inherits: 2.0.4 + readable-stream: 3.6.2 + stream-shift: 1.0.3 + optional: true + eastasianwidth@0.2.0: {} ecc-jsbn@0.1.2: @@ -10604,6 +10943,9 @@ snapshots: etag@1.8.1: {} + event-target-shim@5.0.1: + optional: true + eventemitter3@5.0.1: {} events-universal@1.0.1: @@ -10718,8 +11060,7 @@ snapshots: transitivePeerDependencies: - supports-color - extend@3.0.2: - optional: true + extend@3.0.2: {} extract-zip@1.7.0: dependencies: @@ -10744,6 +11085,8 @@ snapshots: extsprintf@1.3.0: optional: true + farmhash-modern@1.1.0: {} + fast-deep-equal@3.1.3: {} fast-diff@1.3.0: {} @@ -10776,6 +11119,11 @@ snapshots: fast-uri@3.1.0: {} + fast-xml-parser@4.5.3: + dependencies: + strnum: 1.1.2 + optional: true + fast-xml-parser@5.2.5: dependencies: strnum: 2.1.1 @@ -10784,6 +11132,10 @@ snapshots: dependencies: reusify: 1.1.0 + faye-websocket@0.11.4: + dependencies: + websocket-driver: 0.7.4 + fb-watchman@2.0.2: dependencies: bser: 2.1.1 @@ -10850,6 +11202,26 @@ snapshots: locate-path: 6.0.0 path-exists: 4.0.0 + firebase-admin@13.6.0: + dependencies: + '@fastify/busboy': 3.2.0 + '@firebase/database-compat': 2.1.0 + '@firebase/database-types': 1.0.16 + '@types/node': 22.19.0 + farmhash-modern: 1.1.0 + fast-deep-equal: 3.1.3 + google-auth-library: 9.15.1 + jsonwebtoken: 9.0.2 + jwks-rsa: 3.2.0 + node-forge: 1.3.3 + uuid: 11.1.0 + optionalDependencies: + '@google-cloud/firestore': 7.11.6 + '@google-cloud/storage': 7.18.0 + transitivePeerDependencies: + - encoding + - supports-color + flat-cache@3.2.0: dependencies: flatted: 3.3.3 @@ -10913,6 +11285,16 @@ snapshots: mime-types: 2.1.35 optional: true + form-data@2.5.5: + dependencies: + asynckit: 0.4.0 + combined-stream: 1.0.8 + es-set-tostringtag: 2.1.0 + hasown: 2.0.2 + mime-types: 2.1.35 + safe-buffer: 5.2.1 + optional: true + form-data@4.0.4: dependencies: asynckit: 0.4.0 @@ -10980,6 +11362,9 @@ snapshots: hasown: 2.0.2 is-callable: 1.2.7 + functional-red-black-tree@1.0.1: + optional: true + functions-have-names@1.2.3: {} gauge@3.0.2: @@ -10994,6 +11379,26 @@ snapshots: strip-ansi: 6.0.1 wide-align: 1.1.5 + gaxios@6.7.1: + dependencies: + extend: 3.0.2 + https-proxy-agent: 7.0.6 + is-stream: 2.0.1 + node-fetch: 2.7.0 + uuid: 9.0.1 + transitivePeerDependencies: + - encoding + - supports-color + + gcp-metadata@6.1.1: + dependencies: + gaxios: 6.7.1 + google-logging-utils: 0.0.2 + json-bigint: 1.0.0 + transitivePeerDependencies: + - encoding + - supports-color + generate-password@1.7.1: {} gensync@1.0.0-beta.2: {} @@ -11109,12 +11514,53 @@ snapshots: merge2: 1.4.1 slash: 3.0.0 + google-auth-library@9.15.1: + dependencies: + base64-js: 1.5.1 + ecdsa-sig-formatter: 1.0.11 + gaxios: 6.7.1 + gcp-metadata: 6.1.1 + gtoken: 7.1.0 + jws: 4.0.0 + transitivePeerDependencies: + - encoding + - supports-color + + google-gax@4.6.1: + dependencies: + '@grpc/grpc-js': 1.14.0 + '@grpc/proto-loader': 0.7.15 + '@types/long': 4.0.2 + abort-controller: 3.0.0 + duplexify: 4.1.3 + google-auth-library: 9.15.1 + node-fetch: 2.7.0 + object-hash: 3.0.0 + proto3-json-serializer: 2.0.2 + protobufjs: 7.5.4 + retry-request: 7.0.2 + uuid: 9.0.1 + transitivePeerDependencies: + - encoding + - supports-color + optional: true + + google-logging-utils@0.0.2: {} + gopd@1.2.0: {} graceful-fs@4.2.11: {} graphemer@1.4.0: {} + gtoken@7.1.0: + dependencies: + gaxios: 6.7.1 + jws: 4.0.0 + transitivePeerDependencies: + - encoding + - supports-color + handlebars@4.7.8: dependencies: minimist: 1.2.8 @@ -11169,6 +11615,9 @@ snapshots: hpagent@0.1.2: {} + html-entities@2.6.0: + optional: true + html-escaper@2.0.2: {} html-pdf@3.0.1: @@ -11192,6 +11641,17 @@ snapshots: next-line: 1.1.0 optional: true + http-parser-js@0.5.10: {} + + http-proxy-agent@5.0.0: + dependencies: + '@tootallnate/once': 2.0.0 + agent-base: 6.0.2 + debug: 4.4.3 + transitivePeerDependencies: + - supports-color + optional: true + http-proxy-agent@7.0.2: dependencies: agent-base: 7.1.4 @@ -11878,6 +12338,10 @@ snapshots: jsesc@3.1.0: {} + json-bigint@1.0.0: + dependencies: + bignumber.js: 9.3.1 + json-buffer@3.0.1: {} json-parse-even-better-errors@2.3.1: {} @@ -12225,6 +12689,9 @@ snapshots: mime@2.6.0: {} + mime@3.0.0: + optional: true + mimic-fn@2.1.0: {} mimic-fn@4.0.0: {} @@ -12392,6 +12859,8 @@ snapshots: dependencies: whatwg-url: 5.0.0 + node-forge@1.3.3: {} + node-gyp-build-optional-packages@5.2.2: dependencies: detect-libc: 2.1.0 @@ -12824,6 +13293,11 @@ snapshots: kleur: 3.0.3 sisteransi: 1.0.5 + proto3-json-serializer@2.0.2: + dependencies: + protobufjs: 7.5.4 + optional: true + protobufjs@7.5.4: dependencies: '@protobufjs/aspromise': 1.1.2 @@ -13144,6 +13618,16 @@ snapshots: restructure@2.0.1: {} + retry-request@7.0.2: + dependencies: + '@types/request': 2.48.13 + extend: 3.0.2 + teeny-request: 9.0.0 + transitivePeerDependencies: + - encoding + - supports-color + optional: true + retry@0.13.1: {} reusify@1.1.0: {} @@ -13510,6 +13994,14 @@ snapshots: readable-stream: 3.6.2 optional: true + stream-events@1.0.5: + dependencies: + stubs: 3.0.0 + optional: true + + stream-shift@1.0.3: + optional: true + streamsearch@1.1.0: {} streamx@2.23.0: @@ -13588,12 +14080,18 @@ snapshots: strip-json-comments@3.1.1: {} + strnum@1.1.2: + optional: true + strnum@2.1.1: {} strtok3@10.3.4: dependencies: '@tokenizer/token': 0.3.0 + stubs@3.0.0: + optional: true + superagent@7.1.6: dependencies: component-emitter: 1.3.1 @@ -13693,6 +14191,18 @@ snapshots: mkdirp: 1.0.4 yallist: 4.0.0 + teeny-request@9.0.0: + dependencies: + http-proxy-agent: 5.0.0 + https-proxy-agent: 5.0.1 + node-fetch: 2.7.0 + stream-events: 1.0.5 + uuid: 9.0.1 + transitivePeerDependencies: + - encoding + - supports-color + optional: true + telejson@7.2.0: dependencies: memoizerific: 1.11.3 @@ -14160,6 +14670,14 @@ snapshots: - esbuild - uglify-js + websocket-driver@0.7.4: + dependencies: + http-parser-js: 0.5.10 + safe-buffer: 5.2.1 + websocket-extensions: 0.1.4 + + websocket-extensions@0.1.4: {} + whatwg-url@5.0.0: dependencies: tr46: 0.0.3 From e485fba8f02cce50118effd7ef062fcd57c9b6a2 Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Fri, 30 Jan 2026 17:07:45 +0530 Subject: [PATCH 06/13] feat: added firebase dependancy Signed-off-by: Rinkal Bhojani --- package.json | 1 + pnpm-lock.yaml | 3 +++ 2 files changed, 4 insertions(+) diff --git a/package.json b/package.json index 72cc0bfcf..5d16bf99c 100644 --- a/package.json +++ b/package.json @@ -78,6 +78,7 @@ "dotenv": "^16.0.3", "express": "^4.18.2", "express-useragent": "^1.0.15", + "firebase-admin": "^13.6.0", "fs": "0.0.1-security", "generate-password": "^1.7.0", "helmet": "^7.0.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c1143fdc3..0a0bbec1a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -233,6 +233,9 @@ importers: express-useragent: specifier: ^1.0.15 version: 1.0.15 + firebase-admin: + specifier: ^13.6.0 + version: 13.6.0 fs: specifier: 0.0.1-security version: 0.0.1-security From 04e85bdc4ea5b8cec2f60655b78afe4a84abd97b Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Fri, 30 Jan 2026 17:19:29 +0530 Subject: [PATCH 07/13] fix: fixed build error Signed-off-by: Rinkal Bhojani --- apps/notification/src/nats/jetstream.setup.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index 7da124637..eeff66408 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -171,7 +171,7 @@ export async function publishToJetStream( subject, stream: ack.stream, sequence: ack.seq, - dataSize: payload.length, + //dataSize: payload.length, duplicate: ack.duplicate || false })}` ); From fb7114e9b055cef93aae48f4f00a56e4c274820f Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Sun, 1 Feb 2026 23:10:54 +0530 Subject: [PATCH 08/13] fix: corrected NATS servers URL setup logic Signed-off-by: Rinkal Bhojani --- apps/notification/src/nats/nats.service.ts | 2 +- libs/service/nats.options.ts | 6 ++---- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/apps/notification/src/nats/nats.service.ts b/apps/notification/src/nats/nats.service.ts index e2a1527b0..a800b0e25 100644 --- a/apps/notification/src/nats/nats.service.ts +++ b/apps/notification/src/nats/nats.service.ts @@ -18,7 +18,7 @@ export class NatsService implements OnModuleDestroy { this.logger.log('[NATS] starting connection...'); this.nc = await connect({ - servers: [`${process.env.NATS_HOST}:${process.env.NATS_PORT}`], + servers: `${process.env.NATS_URL}`.split(','), authenticator: usernamePasswordAuthenticator(`${process.env.NATS_USER}`, `${process.env.NATS_PASSWORD}`) }); this.js = this.nc.jetstream(); diff --git a/libs/service/nats.options.ts b/libs/service/nats.options.ts index 6051ec0f3..36f717210 100644 --- a/libs/service/nats.options.ts +++ b/libs/service/nats.options.ts @@ -5,14 +5,12 @@ export const commonNatsOptions = (name: string, isClient = true) => { const common: NatsOptions = { transport: Transport.NATS, options: { - url: `nats://${process.env.NATS_HOST}:${process.env.NATS_PORT}`, + servers: `${process.env.NATS_URL}`.split(','), name, maxReconnectAttempts: -1, reconnectTimeWait: 3000 } }; - const result = isClient - ? { ...common, options: { ...common.options, reconnect: true } } - : common; + const result = isClient ? { ...common, options: { ...common.options, reconnect: true } } : common; return result; }; From ea4b4cbff3d53ec1227100f7b0ccd994cdbb7c83 Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Mon, 2 Feb 2026 12:59:49 +0530 Subject: [PATCH 09/13] fix: refactored code and externalized configurational values Signed-off-by: Rinkal Bhojani --- .env.demo | 13 ++ .env.sample | 12 ++ .../src/nats/jetstream.consumer.ts | 43 +++--- apps/notification/src/nats/jetstream.setup.ts | 124 ++++-------------- apps/notification/src/nats/nats-subscriber.ts | 107 +-------------- apps/notification/src/nats/nats.module.ts | 4 +- apps/notification/src/nats/nats.service.ts | 13 +- 7 files changed, 89 insertions(+), 227 deletions(-) diff --git a/.env.demo b/.env.demo index 4b8032d5e..adf5531bf 100644 --- a/.env.demo +++ b/.env.demo @@ -233,3 +233,16 @@ PRISMA_LOGS = error # DB_ALERT_ENABLE= # DB_ALERT_EMAILS= + + + +PROJECT_ID= +CLIENT_EMAIL= +PRIVATE_KEY= + +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; \ No newline at end of file diff --git a/.env.sample b/.env.sample index 9eedf47d2..1d30dfc78 100644 --- a/.env.sample +++ b/.env.sample @@ -262,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= + + +PROJECT_ID= +CLIENT_EMAIL= +PRIVATE_KEY= + +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; \ No newline at end of file diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index 80498c57b..77a1b85dc 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -7,13 +7,16 @@ import { ensureDidStream, publishToJetStream, PULL_CONSUMER, - STREAM + 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 { EVENT_USER_ACK } from './nats-subscriber'; +const EVENT_PRESENTATION_ACK = 'presentation.ack'; +const EVENT_PRESENTATION_PURGED = 'presentation.purged'; @Injectable() export class JetStreamConsumer implements OnApplicationBootstrap { constructor( @@ -31,7 +34,6 @@ export class JetStreamConsumer implements OnApplicationBootstrap { clientEmail: process.env.CLIENT_EMAIL, privateKey: process.env.PRIVATE_KEY.replace(/\\n/g, '\n') }) - // credential: admin.credential.cert('/home/rinkal-bhojani/Rinkal/platform/dyuat-80c22-firebase-adminsdk-41x7t-3e8d2469ee.json') }); } @@ -39,14 +41,7 @@ export class JetStreamConsumer implements OnApplicationBootstrap { const jsm = this.nats.jetstreamManager(); await ensureConsumer(jsm); - const consumer: Consumer = await js.consumers.get(STREAM, PULL_CONSUMER); - // const consumer: Consumer = await js.consumers.get( - // STREAM, - // { - // // name_prefix: CONSUMER, - // filterSubjects: ['hubStream.presentation.test.blr'] - // } - // ); + const consumer: Consumer = await js.consumers.get(AGGREGATE_STREAM, PULL_CONSUMER); this.logger.log('[NATS] JetStream consumer started'); @@ -73,9 +68,8 @@ export class JetStreamConsumer implements OnApplicationBootstrap { const payload = msg.json(); const notificationDetail = await this.holderNotificationRepository.getHolderNotificationBySessionId(sessionId); - this.logger.log(`[NATS] Message received ${JSON.stringify(payload)}`); + this.logger.debug(`[NATS] Message received ${JSON.stringify(payload)}`); this.logger.log(`[NATS] Processing message, ${JSON.stringify({ deliveryCount: msg.info.deliveryCount })}`); - this.logger.log(`[NATS] Consdition....', ${JSON.stringify({ IsFCMSatisfied: 3 < msg.info.deliveryCount })}`); if (3 < msg.info.deliveryCount) { //------------ Moving message to DID stream ---------------// this.logger.log('[NATS] Moving message to DID stream for FCM notification'); @@ -95,16 +89,16 @@ export class JetStreamConsumer implements OnApplicationBootstrap { //------------- Sending Push Notification via FCM ----------------// let notificationTitle = ''; - if ('presentation.ack' === event) { + if (EVENT_PRESENTATION_ACK === event) { notificationTitle = `Your data delivered to ${orgCode}`; - } else if ('presentation.purged' === event) { + } 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 the app to view more details` + body: `Open an app to view more details` }, token: notificationDetail.fcmToken }; @@ -130,13 +124,18 @@ export class JetStreamConsumer implements OnApplicationBootstrap { msg.nak(); return; } - await this.nats.publish(`${notificationDetail.holderDid}`, { - //`${notificationDetail.holderDid}`, { - payload, - ackKey, // opaque token user must return - subject: msg.subject, - event: `${domain}.${event}` - }); + await this.nats.publish( + `${notificationDetail.holderDid}`, + { + //`${notificationDetail.holderDid}`, { + payload, + ackKey, // opaque token user must return + subject: msg.subject, + msg, + event: `${domain}.${event}` + }, + { reply: EVENT_USER_ACK } + ); // await this.nats.publish('did-key', payload); // ACK reply to APP this.logger.log(`[NATS] Message published to ${notificationDetail.holderDid} for session ${sessionId}`); diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index eeff66408..a9265e745 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -1,23 +1,11 @@ /* eslint-disable camelcase */ import { Logger } from '@nestjs/common'; -import { - AckPolicy, - DeliverPolicy, - JetStreamClient, - JetStreamManager, - ReplayPolicy, - RetentionPolicy, - StorageType, - nanos -} from 'nats'; +import { AckPolicy, DeliverPolicy, JetStreamClient, JetStreamManager, RetentionPolicy, StorageType, nanos } from 'nats'; import assert = require('node:assert'); -export const SOURCE_STREAM = 'notify'; -export const STREAM = 'aggregate'; -export const DID_STREAM = 'did-notify'; -export const SUBJECTS = ['aggregate.>']; -export const CONSUMER = 'hub-consumer'; -export const PULL_CONSUMER = 'hub-pull-consumer'; +export const AGGREGATE_STREAM = process.env.AGGREGATE_STREAM || 'aggregate'; +export const DID_STREAM = process.env.DID_STREAM || 'did-notify'; +export const PULL_CONSUMER = process.env.PULL_CONSUMER || 'hub-pull-consumer'; const logger = new Logger('JetStreamSetup'); /** @@ -25,30 +13,31 @@ const logger = new Logger('JetStreamSetup'); */ export async function ensureStream(jsm: JetStreamManager): Promise { try { - // await jsm.streams.delete(STREAM); - await jsm.streams.info(STREAM); - logger.log(`[NATS] Stream '${STREAM}' already exists`); - } catch { - logger.log(`[NATS] Creating stream '${STREAM}'`); + await jsm.streams.info(AGGREGATE_STREAM); + logger.log(`[NATS] Stream '${AGGREGATE_STREAM}' already exists`); + } catch (error) { + logger.error( + `[NATS] Error checking stream '${AGGREGATE_STREAM}': ${error instanceof Error ? error.message : 'Unknown error'}` + ); + logger.log(`[NATS] Creating stream '${AGGREGATE_STREAM}'`); await jsm.streams.add({ - name: STREAM, + name: AGGREGATE_STREAM, //subjects: SUBJECTS, retention: RetentionPolicy.Workqueue, storage: StorageType.File, - sources: [ - { name: SOURCE_STREAM, domain: 'del' }, - { name: SOURCE_STREAM, domain: 'blr' } - ] + sources: [] }); } } export async function ensureDidStream(jsm: JetStreamManager): Promise { try { - // await jsm.streams.delete(DID_STREAM); await jsm.streams.info(DID_STREAM); logger.log(`[NATS] Stream '${DID_STREAM}' already exists`); - } catch { + } catch (error) { + logger.error( + `[NATS] Error checking stream '${DID_STREAM}': ${error instanceof Error ? error.message : 'Unknown error'}` + ); logger.log(`[NATS] Creating stream '${DID_STREAM}'`); await jsm.streams.add({ name: DID_STREAM, @@ -64,79 +53,23 @@ export async function ensureDidStream(jsm: JetStreamManager): Promise { */ export async function ensureConsumer(jsm: JetStreamManager): Promise { try { - // await jsm.consumers.delete(STREAM, CONSUMER); - // await jsm.consumers.update(STREAM, CONSUMER, { - // // eslint-disable-next-line camelcase - // // filter_subject: '*.presentation.test.blr' - // }); - await jsm.consumers.info(STREAM, PULL_CONSUMER); + await jsm.consumers.info(AGGREGATE_STREAM, PULL_CONSUMER); logger.log(`[NATS] Consumer '${PULL_CONSUMER}' already exists`); - } catch { - logger.log(`[NATS] Creating consumer '${CONSUMER}'`); - await jsm.consumers.add(STREAM, { - name: PULL_CONSUMER, - durable_name: PULL_CONSUMER, - ack_policy: AckPolicy.Explicit, - deliver_policy: DeliverPolicy.All, - ack_wait: nanos(10_000), - max_deliver: 4 - // ,filter_subjects: ['hubStream.presentation.test.*'] - }); - } -} - -export async function ensureSessionConsumer(jsm: JetStreamManager, sessionId: string): Promise { - const consumerName = `notify-session-${sessionId}`; - - try { - // await jsm.consumers.delete(STREAM, consumerName); - await jsm.consumers.info(STREAM, consumerName); - logger.log(`[NATS] Consumer '${consumerName}' already exists`); - return consumerName; } catch (error) { logger.error( - `Error creatingg consumer info for '${consumerName}': ${error instanceof Error ? error.message : 'Unknown error'}` + `[NATS] Error checking consumer '${PULL_CONSUMER}': ${error instanceof Error ? error.message : 'Unknown error'}` ); - logger.log(`[NATS] Creating consumer '${consumerName}'`); - await jsm.consumers.add(STREAM, { - name: consumerName, - // durable_name: consumerName, - ack_policy: AckPolicy.Explicit, - deliver_policy: DeliverPolicy.All, - filter_subject: `notify.*.*.*.${sessionId}`, - replay_policy: ReplayPolicy.Instant, - // max_ack_pending: 1000, - inactive_threshold: 30_000 - }); - - return consumerName; - } -} - -export async function ensureSessionConsumerWithDidStream(jsm: JetStreamManager, sessionId: string): Promise { - const consumerName = `notify-session-${sessionId}`; - - try { - await jsm.consumers.info(DID_STREAM, consumerName); - logger.log(`[NATS] Consumer '${consumerName}' already exists`); - return consumerName; - } catch (error) { - logger.error( - `Error creatingg consumer info for '${consumerName}': ${error instanceof Error ? error.message : 'Unknown error'}` - ); - logger.log(`[NATS] Creating consumer '${consumerName}'`); - await jsm.consumers.add(DID_STREAM, { - name: consumerName, - // durable_name: consumerName, + logger.log(`[NATS] Creating consumer '${PULL_CONSUMER}'`); + await jsm.consumers.add(AGGREGATE_STREAM, { + name: PULL_CONSUMER, + durable_name: PULL_CONSUMER, ack_policy: AckPolicy.Explicit, deliver_policy: DeliverPolicy.All, - filter_subject: `notify.*.*.*.${sessionId}`, - replay_policy: ReplayPolicy.Instant, - // max_ack_pending: 1000, - inactive_threshold: 30_000 + ack_wait: process.env.CONSUMER_CONFIG_ACK_WAIT + ? nanos(Number(process.env.CONSUMER_CONFIG_ACK_WAIT)) + : nanos(10_000), + max_deliver: process.env.CONSUMER_CONFIG_MAX_DELIVER ? Number(process.env.CONSUMER_CONFIG_MAX_DELIVER) : 4 }); - - return consumerName; } } @@ -171,7 +104,6 @@ export async function publishToJetStream( subject, stream: ack.stream, sequence: ack.seq, - //dataSize: payload.length, duplicate: ack.duplicate || false })}` ); @@ -181,7 +113,7 @@ export async function publishToJetStream( seq: ack.seq }; } catch (error) { - logger.error(`------------${JSON.stringify(error)}`); + logger.error(`[NATS] Error publishing to JetStream: ${JSON.stringify(error)}`); logger.error('[NATS] JetStream publish failed', { subject, error: error instanceof Error ? error.message : 'Unknown error' diff --git a/apps/notification/src/nats/nats-subscriber.ts b/apps/notification/src/nats/nats-subscriber.ts index 9071abd78..ae89a7b8f 100644 --- a/apps/notification/src/nats/nats-subscriber.ts +++ b/apps/notification/src/nats/nats-subscriber.ts @@ -1,13 +1,11 @@ import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { NatsService } from './nats.service'; -import { INatsUserRequestData } from './interfaces'; -import { DID_STREAM, ensureSessionConsumer, ensureSessionConsumerWithDidStream, STREAM } from './jetstream.setup'; -import { JetStreamClient } from 'nats'; import { PendingAckStore } from './pendingAckStore'; const sd = new TextDecoder(); +export const EVENT_USER_ACK = 'user.ack'; @Injectable() -export class TestSubsciber implements OnApplicationBootstrap { +export class NotificationNATSSubsciber implements OnApplicationBootstrap { constructor( private readonly nats: NatsService, private readonly logger: Logger, @@ -15,12 +13,7 @@ export class TestSubsciber implements OnApplicationBootstrap { ) {} async onApplicationBootstrap(): Promise { - this.subscribe('active.user', async (data) => { - this.logger.log(`[NATS] activate.user Subsciber received data: ${JSON.stringify(data)}`); - await this.activateUserConsumers(data as INatsUserRequestData); - }); - - this.subscribe('user.ack', async (data) => { + this.subscribe(EVENT_USER_ACK, async (data) => { this.logger.log('[NATS] user.ack Subsciber received data:', data); await this.handleUserAck(data as { ackKey: string }); }); @@ -64,50 +57,6 @@ export class TestSubsciber implements OnApplicationBootstrap { } } - private async activateUserConsumers(data: INatsUserRequestData): Promise { - const { did, sessions } = data; - - const js = this.nats.jetstream(); - const jsm = this.nats.jetstreamManager(); - this.logger.log(`[NATS] Started activating consumer for user DID: ${did} with sessions count: ${sessions.length}`); - - // Create all consumers in parallel first (dependency) - await Promise.all(sessions.map((session) => ensureSessionConsumer(jsm, session.sessionId))); - - // After all consumers are created, consume messages from all sessions in parallel - await Promise.all(sessions.map((session) => this.consumeSessionMessages(js, session.sessionId, did))); - } - - private async consumeSessionMessages(js: JetStreamClient, sessionId: string, did: string): Promise { - const consumerName = `notify-session-${sessionId}`; - this.logger.log(`[NATS] Getting consumer ${consumerName} to fetch messages for session ${sessionId}`); - const consumer = await js.consumers.get(STREAM, consumerName); - - const iter = await consumer.fetch({ - // eslint-disable-next-line camelcase - max_messages: 10 - }); - this.logger.log(`[NATS] Started fetching messages for consumer ${consumerName}`); - for await (const msg of iter) { - const payload = msg.json(); - // eslint-disable-next-line @typescript-eslint/no-unused-vars - const [_, domain, event, orgCode, sessionId] = msg.subject.split('.'); - - const ackKey = this.pendingAckStore.save('notify', consumerName, msg); - - await this.nats.publish(`${did}`, { - payload, - ackKey, // opaque token user must return - subject: msg.subject, - event: `${domain}.${event}` - }); - this.logger.log(`[NATS] Published message to user DID ${did} for session ${sessionId}`); - } - this.logger.log(`[NATS] Finished processing messages for session ${sessionId}, deleting consumer ${consumerName}`); - await this.nats.jetstreamManager().consumers.delete(STREAM, consumerName); - this.logger.log(`[NATS] Deleted consumer ${consumerName} after processing messages for session ${sessionId}`); - } - async handleUserAck(data: { ackKey: string }): Promise { const msg = this.pendingAckStore.get(data.ackKey); @@ -124,54 +73,4 @@ export class TestSubsciber implements OnApplicationBootstrap { this.logger.error(`[NATS] Failed to ACK message ${JSON.stringify(err)}`); } } - - // ------ For did-notify stream testing ------ // - - private async activateUserConsumersWithDidNotify(data: INatsUserRequestData): Promise { - const { did, sessions } = data; - - const js = this.nats.jetstream(); - const jsm = this.nats.jetstreamManager(); - this.logger.log(`[NATS] Started activating consumer for user DID: ${did} with sessions count: ${sessions.length}`); - - // Create all consumers in parallel first (dependency) - await Promise.all(sessions.map((session) => ensureSessionConsumerWithDidStream(jsm, session.sessionId))); - - // After all consumers are created, consume messages from all sessions in parallel - await Promise.all(sessions.map((session) => this.consumeSessionMessagesWithDidNotify(js, session.sessionId, did))); - } - - private async consumeSessionMessagesWithDidNotify( - js: JetStreamClient, - sessionId: string, - did: string - ): Promise { - const consumerName = `notify-session-${sessionId}`; - this.logger.log(`[NATS] Getting consumer ${consumerName} to fetch messages for session ${sessionId}`); - const consumer = await js.consumers.get(DID_STREAM, consumerName); - - const iter = await consumer.fetch({ - // eslint-disable-next-line camelcase - max_messages: 10 - }); - this.logger.log(`[NATS] Started fetching messages for consumer ${consumerName}`); - for await (const msg of iter) { - const payload = msg.json(); - // eslint-disable-next-line @typescript-eslint/no-unused-vars - const [_, domain, event, orgCode, sessionId] = msg.subject.split('.'); - - const ackKey = this.pendingAckStore.save('notify', consumerName, msg); - - await this.nats.publish(`${did}`, { - payload, - ackKey, // opaque token user must return - subject: msg.subject, - event: `${domain}.${event}` - }); - this.logger.log(`[NATS] Published message to user DID ${did} for session ${sessionId}`); - } - this.logger.log(`[NATS] Finished processing messages for session ${sessionId}, deleting consumer ${consumerName}`); - await this.nats.jetstreamManager().consumers.delete(STREAM, consumerName); - this.logger.log(`[NATS] Deleted consumer ${consumerName} after processing messages for session ${sessionId}`); - } } diff --git a/apps/notification/src/nats/nats.module.ts b/apps/notification/src/nats/nats.module.ts index a3c04d74f..a2588e702 100644 --- a/apps/notification/src/nats/nats.module.ts +++ b/apps/notification/src/nats/nats.module.ts @@ -2,7 +2,7 @@ import { Logger, Module, OnModuleInit } from '@nestjs/common'; import { NatsService } from './nats.service'; import { JetStreamConsumer } from './jetstream.consumer'; import { ensureStream } from './jetstream.setup'; -import { TestSubsciber } from './nats-subscriber'; +import { NotificationNATSSubsciber } from './nats-subscriber'; import { ConfigModule } from '@nestjs/config'; import { CommonModule } from '@credebl/common'; import { GlobalConfigModule } from '@credebl/config'; @@ -28,7 +28,7 @@ import { PrismaService } from '@credebl/prisma-service'; providers: [ NatsService, JetStreamConsumer, - TestSubsciber, + NotificationNATSSubsciber, Logger, PendingAckStore, HolderNotificationRepository, diff --git a/apps/notification/src/nats/nats.service.ts b/apps/notification/src/nats/nats.service.ts index a800b0e25..4d5e8400b 100644 --- a/apps/notification/src/nats/nats.service.ts +++ b/apps/notification/src/nats/nats.service.ts @@ -1,5 +1,12 @@ import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common'; -import { connect, JetStreamClient, JetStreamManager, NatsConnection, usernamePasswordAuthenticator } from 'nats'; +import { + connect, + JetStreamClient, + JetStreamManager, + NatsConnection, + PublishOptions, + usernamePasswordAuthenticator +} from 'nats'; @Injectable() export class NatsService implements OnModuleDestroy { @@ -49,10 +56,10 @@ export class NatsService implements OnModuleDestroy { } } - async publish(subject: string, payload: unknown): Promise { + async publish(subject: string, payload: unknown, options: PublishOptions): Promise { if (!this.connected) { throw new Error('NATS not connected yet'); } - this.nc.publish(subject, Buffer.from(JSON.stringify(payload)), { reply: 'ack-notification' }); + this.nc.publish(subject, Buffer.from(JSON.stringify(payload)), options); } } From 52befa84d4dab6e25d88f2fc554bef71970372ad Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Mon, 2 Feb 2026 19:13:15 +0530 Subject: [PATCH 10/13] fix: minor fixes Signed-off-by: Rinkal Bhojani --- .env.demo | 6 +++--- .env.sample | 6 +++--- apps/api-gateway/src/main.ts | 1 + apps/notification/src/nats/jetstream.consumer.ts | 11 +++++------ apps/notification/src/nats/jetstream.setup.ts | 3 +-- 5 files changed, 13 insertions(+), 14 deletions(-) diff --git a/.env.demo b/.env.demo index adf5531bf..2136ef235 100644 --- a/.env.demo +++ b/.env.demo @@ -243,6 +243,6 @@ PRIVATE_KEY= 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; \ No newline at end of file +AGGREGATE_STREAM=aggregate +DID_STREAM=did-notify +PULL_CONSUMER=hub-pull-consumer \ No newline at end of file diff --git a/.env.sample b/.env.sample index 1d30dfc78..6c02be764 100644 --- a/.env.sample +++ b/.env.sample @@ -271,6 +271,6 @@ PRIVATE_KEY= 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; \ No newline at end of file +AGGREGATE_STREAM=aggregate +DID_STREAM=did-notify +PULL_CONSUMER=hub-pull-consumer \ No newline at end of file diff --git a/apps/api-gateway/src/main.ts b/apps/api-gateway/src/main.ts index 8fb86c465..0c4dcef47 100644 --- a/apps/api-gateway/src/main.ts +++ b/apps/api-gateway/src/main.ts @@ -125,6 +125,7 @@ async function bootstrap(): Promise { 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'); diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index 77a1b85dc..20125a28f 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -51,10 +51,11 @@ export class JetStreamConsumer implements OnApplicationBootstrap { } private async consume(consumer: Consumer): Promise { + 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); + 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}`; @@ -118,20 +119,18 @@ export class JetStreamConsumer implements OnApplicationBootstrap { const ackKey = this.pendingAckStore.save('notify', consumerName, msg); - this.logger.log('[NATS] Notification detail fetched', notificationDetail); + this.logger.log(`[NATS] Notification detail fetched: ${JSON.stringify(notificationDetail)}`); if (!notificationDetail) { - this.logger.error('[NATS] No notification detail found for session ID:', sessionId); + this.logger.error(`[NATS] No notification detail found for session ID: ${sessionId}`); msg.nak(); return; } await this.nats.publish( `${notificationDetail.holderDid}`, { - //`${notificationDetail.holderDid}`, { payload, - ackKey, // opaque token user must return + ackKey, subject: msg.subject, - msg, event: `${domain}.${event}` }, { reply: EVENT_USER_ACK } diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index a9265e745..42a892b18 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -6,7 +6,6 @@ import assert = require('node:assert'); export const AGGREGATE_STREAM = process.env.AGGREGATE_STREAM || 'aggregate'; export const DID_STREAM = process.env.DID_STREAM || 'did-notify'; export const PULL_CONSUMER = process.env.PULL_CONSUMER || 'hub-pull-consumer'; - const logger = new Logger('JetStreamSetup'); /** * Ensures the JetStream stream exists @@ -100,7 +99,7 @@ export async function publishToJetStream( ); logger.log( - `[NATS] Message published to JetStream, ${JSON.stringify({ + `[NATS] Message published to JetStream - ${JSON.stringify({ subject, stream: ack.stream, sequence: ack.seq, From 321e622b82bea74fba84009d878c2f8d428da3fe Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Mon, 2 Feb 2026 20:05:32 +0530 Subject: [PATCH 11/13] fix: minor fixes Signed-off-by: Rinkal Bhojani --- .../src/nats/jetstream.consumer.ts | 20 +++++++------------ apps/notification/src/nats/nats.service.ts | 13 +++--------- 2 files changed, 10 insertions(+), 23 deletions(-) diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index 20125a28f..d0aa5b546 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -13,7 +13,6 @@ 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 { EVENT_USER_ACK } from './nats-subscriber'; const EVENT_PRESENTATION_ACK = 'presentation.ack'; const EVENT_PRESENTATION_PURGED = 'presentation.purged'; @@ -119,23 +118,18 @@ export class JetStreamConsumer implements OnApplicationBootstrap { const ackKey = this.pendingAckStore.save('notify', consumerName, msg); - this.logger.log(`[NATS] Notification detail fetched: ${JSON.stringify(notificationDetail)}`); + 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(); return; } - await this.nats.publish( - `${notificationDetail.holderDid}`, - { - payload, - ackKey, - subject: msg.subject, - event: `${domain}.${event}` - }, - { reply: EVENT_USER_ACK } - ); - // await this.nats.publish('did-key', payload); // ACK reply to APP + 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}`); } diff --git a/apps/notification/src/nats/nats.service.ts b/apps/notification/src/nats/nats.service.ts index 4d5e8400b..7c16a54e0 100644 --- a/apps/notification/src/nats/nats.service.ts +++ b/apps/notification/src/nats/nats.service.ts @@ -1,12 +1,5 @@ import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common'; -import { - connect, - JetStreamClient, - JetStreamManager, - NatsConnection, - PublishOptions, - usernamePasswordAuthenticator -} from 'nats'; +import { connect, JetStreamClient, JetStreamManager, NatsConnection, usernamePasswordAuthenticator } from 'nats'; @Injectable() export class NatsService implements OnModuleDestroy { @@ -56,10 +49,10 @@ export class NatsService implements OnModuleDestroy { } } - async publish(subject: string, payload: unknown, options: PublishOptions): Promise { + async publish(subject: string, payload: unknown): Promise { if (!this.connected) { throw new Error('NATS not connected yet'); } - this.nc.publish(subject, Buffer.from(JSON.stringify(payload)), options); + this.nc.publish(subject, Buffer.from(JSON.stringify(payload))); } } From 71bccd38ed7eef5821515aaaabbd2cc273a93f46 Mon Sep 17 00:00:00 2001 From: shitrerohit Date: Tue, 3 Feb 2026 19:57:50 +0530 Subject: [PATCH 12/13] fix:minor issues Signed-off-by: shitrerohit --- .env.demo | 4 ++-- .env.sample | 4 ++-- apps/notification/src/nats/jetstream.consumer.ts | 13 ++++++++++--- apps/notification/src/nats/jetstream.setup.ts | 3 ++- apps/notification/src/nats/nats-subscriber.ts | 4 ++-- apps/notification/src/nats/pendingAckStore.ts | 5 ++++- libs/service/nats.options.ts | 12 +++++++++++- 7 files changed, 33 insertions(+), 12 deletions(-) diff --git a/.env.demo b/.env.demo index 2136ef235..67e794f54 100644 --- a/.env.demo +++ b/.env.demo @@ -26,8 +26,8 @@ NATS_HOST=your-ip NATS_PORT=4222 NATS_URL=nats://your-ip:4222 -NATS_USER=xxxx NATS_PASSWORD=xxxx +NATS_USER=xxxx REDIS_HOST=your-ip REDIS_PORT=6379 @@ -236,9 +236,9 @@ PRISMA_LOGS = error -PROJECT_ID= CLIENT_EMAIL= PRIVATE_KEY= +PROJECT_ID= CONSUMER_CONFIG_ACK_WAIT=10_000 # IN nanos(10_000) CONSUMER_CONFIG_MAX_DELIVER=4 diff --git a/.env.sample b/.env.sample index 6c02be764..386ddadf0 100644 --- a/.env.sample +++ b/.env.sample @@ -59,8 +59,8 @@ NATS_HOST='0.0.0.0' NATS_PORT=4222 NATS_URL=nats://0.0.0.0:4222 -NATS_USER=xxxx NATS_PASSWORD=xxxx +NATS_USER=xxxx REDIS_HOST='0.0.0.0' REDIS_PORT=6379 @@ -264,9 +264,9 @@ RESEND_API_KEY=re_xxxxxxxxxx # DB_ALERT_ENABLE= -PROJECT_ID= CLIENT_EMAIL= PRIVATE_KEY= +PROJECT_ID= CONSUMER_CONFIG_ACK_WAIT=10_000 # IN nanos(10_000) CONSUMER_CONFIG_MAX_DELIVER=4 diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index d0aa5b546..8e55b2f31 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -27,11 +27,18 @@ export class JetStreamConsumer implements OnApplicationBootstrap { async onApplicationBootstrap(): Promise { 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: process.env.PROJECT_ID, - clientEmail: process.env.CLIENT_EMAIL, - privateKey: process.env.PRIVATE_KEY.replace(/\\n/g, '\n') + projectId, + clientEmail, + privateKey: privateKey.replace(/\\n/g, '\n') }) }); } diff --git a/apps/notification/src/nats/jetstream.setup.ts b/apps/notification/src/nats/jetstream.setup.ts index 42a892b18..8df06230f 100644 --- a/apps/notification/src/nats/jetstream.setup.ts +++ b/apps/notification/src/nats/jetstream.setup.ts @@ -91,7 +91,8 @@ export async function publishToJetStream( try { // const js = await getJetStreamClient(); - logger.log(`------------Publishing to JetStream------------ ${subject} with payload: ${JSON.stringify(payload)}`); + logger.log(`[NATS] Publishing to JetStream: ${subject}`); + logger.debug(`[NATS] Payload for ${subject}:`, { payload }); // Only in debug mode const ack = await jsc.publish( `${subject}`, // sc.encode(JSON.stringify(payload)) diff --git a/apps/notification/src/nats/nats-subscriber.ts b/apps/notification/src/nats/nats-subscriber.ts index ae89a7b8f..c3183db4a 100644 --- a/apps/notification/src/nats/nats-subscriber.ts +++ b/apps/notification/src/nats/nats-subscriber.ts @@ -13,8 +13,8 @@ export class NotificationNATSSubsciber implements OnApplicationBootstrap { ) {} async onApplicationBootstrap(): Promise { - this.subscribe(EVENT_USER_ACK, async (data) => { - this.logger.log('[NATS] user.ack Subsciber received data:', data); + await this.subscribe(EVENT_USER_ACK, async (data) => { + this.logger.log('[NATS] user.ack Subscriber received data:', data); await this.handleUserAck(data as { ackKey: string }); }); } diff --git a/apps/notification/src/nats/pendingAckStore.ts b/apps/notification/src/nats/pendingAckStore.ts index 3179a07a9..75e68caec 100644 --- a/apps/notification/src/nats/pendingAckStore.ts +++ b/apps/notification/src/nats/pendingAckStore.ts @@ -17,7 +17,10 @@ export class PendingAckStore { } save(stream: string, consumer: string, msg: JsMsg): PendingKey { - const info = msg.info!; + const { info } = msg; + if (!info) { + throw new Error('Cannot save message without info metadata'); + } const key = this.makeKey(stream, consumer, info.streamSequence); this.store.set(key, { diff --git a/libs/service/nats.options.ts b/libs/service/nats.options.ts index 36f717210..83918da00 100644 --- a/libs/service/nats.options.ts +++ b/libs/service/nats.options.ts @@ -5,7 +5,17 @@ export const commonNatsOptions = (name: string, isClient = true) => { const common: NatsOptions = { transport: Transport.NATS, options: { - servers: `${process.env.NATS_URL}`.split(','), + servers: (() => { + const raw = process.env.NATS_URL ?? ''; + const servers = raw + .split(',') + .map((s) => s.trim()) + .filter(Boolean); + if (0 === servers.length) { + throw new Error('NATS_URL is required and must contain at least one server'); + } + return servers; + })(), name, maxReconnectAttempts: -1, reconnectTimeWait: 3000 From e250531494b8ee12ca94096feb12396f866d6d66 Mon Sep 17 00:00:00 2001 From: Rinkal Bhojani Date: Wed, 4 Feb 2026 11:16:00 +0530 Subject: [PATCH 13/13] fix: fixed coderabitai issues Signed-off-by: Rinkal Bhojani --- .../src/authz/guards/client-access-guard.ts | 2 +- .../src/nats/jetstream.consumer.ts | 94 +++++++++++-------- apps/notification/src/nats/nats-subscriber.ts | 2 +- apps/notification/src/nats/nats.module.ts | 4 +- apps/notification/src/nats/nats.service.ts | 7 +- .../src/oid4vc-verification.service.ts | 6 +- 6 files changed, 65 insertions(+), 50 deletions(-) diff --git a/apps/api-gateway/src/authz/guards/client-access-guard.ts b/apps/api-gateway/src/authz/guards/client-access-guard.ts index de9390cbf..cac92acdd 100644 --- a/apps/api-gateway/src/authz/guards/client-access-guard.ts +++ b/apps/api-gateway/src/authz/guards/client-access-guard.ts @@ -8,7 +8,7 @@ export class ClientAccessGuard implements CanActivate { const { user } = request; - if (!user.hasOwnProperty('client_id')) { + if (!user || !Object.prototype.hasOwnProperty.call(user, 'client_id')) { throw new UnauthorizedException('You do not have access'); } diff --git a/apps/notification/src/nats/jetstream.consumer.ts b/apps/notification/src/nats/jetstream.consumer.ts index 8e55b2f31..3bf3c1848 100644 --- a/apps/notification/src/nats/jetstream.consumer.ts +++ b/apps/notification/src/nats/jetstream.consumer.ts @@ -13,6 +13,7 @@ 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'; @@ -77,59 +78,26 @@ export class JetStreamConsumer implements OnApplicationBootstrap { 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 })}`); - if (3 < 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 ---------------// - 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)}`); + await this.moveMsgToDidStream(notificationDetail, msg); //------------- Sending Push Notification via FCM ----------------// - 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); - }); + this.sendNotificationToHolder(event, orgCode, notificationDetail); msg.ack(); } else { // ------------- Publishing Messages via NATS ----------------// - const ackKey = this.pendingAckStore.save('notify', consumerName, msg); + 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(); - return; + continue; } await this.nats.publish(`${notificationDetail.holderDid}`, { payload, @@ -149,4 +117,48 @@ export class JetStreamConsumer implements OnApplicationBootstrap { } } } + + 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 { + 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)}`); + } } diff --git a/apps/notification/src/nats/nats-subscriber.ts b/apps/notification/src/nats/nats-subscriber.ts index c3183db4a..9c5c88ff3 100644 --- a/apps/notification/src/nats/nats-subscriber.ts +++ b/apps/notification/src/nats/nats-subscriber.ts @@ -22,7 +22,7 @@ export class NotificationNATSSubsciber implements OnApplicationBootstrap { private async subscribe(subject: string, handler: (data: unknown) => void | Promise): Promise { try { this.logger.log(`[NATS] Setting up subscription ${JSON.stringify(subject)}`); - const nc = await this.nats.nc; + const nc = await this.nats.connect(); const subscription = nc.subscribe(subject); // eslint-disable-next-line @typescript-eslint/explicit-function-return-type diff --git a/apps/notification/src/nats/nats.module.ts b/apps/notification/src/nats/nats.module.ts index a2588e702..46dbf209f 100644 --- a/apps/notification/src/nats/nats.module.ts +++ b/apps/notification/src/nats/nats.module.ts @@ -9,7 +9,6 @@ import { GlobalConfigModule } from '@credebl/config'; import { ConfigModule as PlatformConfig } from '@credebl/config/config.module'; import { LoggerModule } from '@credebl/logger'; import { ContextInterceptorModule } from '@credebl/context'; -import { CacheModule } from '@nestjs/cache-manager'; import { PendingAckStore } from './pendingAckStore'; import { HolderNotificationRepository } from '../holder-notification.repository'; import { PrismaService } from '@credebl/prisma-service'; @@ -22,8 +21,7 @@ import { PrismaService } from '@credebl/prisma-service'; GlobalConfigModule, LoggerModule, PlatformConfig, - ContextInterceptorModule, - CacheModule.register({ host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }) + ContextInterceptorModule ], providers: [ NatsService, diff --git a/apps/notification/src/nats/nats.service.ts b/apps/notification/src/nats/nats.service.ts index 7c16a54e0..61d77bffc 100644 --- a/apps/notification/src/nats/nats.service.ts +++ b/apps/notification/src/nats/nats.service.ts @@ -12,11 +12,16 @@ export class NatsService implements OnModuleDestroy { async connect(): Promise { if (this.connected) { - return; + return this.nc; } this.logger.log('[NATS] starting connection...'); + const { NATS_URL, NATS_USER, NATS_PASSWORD } = process.env; + if (!NATS_URL || !NATS_USER || !NATS_PASSWORD) { + throw new Error('Missing NATS connection env vars (NATS_URL, NATS_USER, NATS_PASSWORD)'); + } + this.nc = await connect({ servers: `${process.env.NATS_URL}`.split(','), authenticator: usernamePasswordAuthenticator(`${process.env.NATS_USER}`, `${process.env.NATS_PASSWORD}`) diff --git a/apps/oid4vc-verification/src/oid4vc-verification.service.ts b/apps/oid4vc-verification/src/oid4vc-verification.service.ts index 6ff6d4ffb..1884abb7f 100644 --- a/apps/oid4vc-verification/src/oid4vc-verification.service.ts +++ b/apps/oid4vc-verification/src/oid4vc-verification.service.ts @@ -322,7 +322,7 @@ export class Oid4vpVerificationService extends BaseService { // Fetch intent template using utilities service this.logger.debug( - `[createVerificationPresentation] fetching intent template for intent=${intent}, orgId=${orgId}` + `[createIntentBasedVerificationPresentation] fetching intent template for intent=${intent}, orgId=${orgId}` ); const templateData = await this.natsClient.sendNatsMessage( this.oid4vpVerificationServiceProxy, @@ -335,7 +335,7 @@ export class Oid4vpVerificationService extends BaseService { } this.logger.debug( - `[createVerificationPresentation] template fetched successfully: ${JSON.stringify(templateData)}` + `[createIntentBasedVerificationPresentation] template fetched successfully: ${JSON.stringify(templateData)}` ); // Build session request using fetched template @@ -393,7 +393,7 @@ export class Oid4vpVerificationService extends BaseService { return createdSession; } catch (error) { this.logger.error( - `[createVerificationPresentation] - error creating verification presentation: ${JSON.stringify(error)}` + `[createIntentBasedVerificationPresentation] - error creating verification presentation: ${JSON.stringify(error)}` ); throw new RpcException(error?.response ?? error); }