From 4397b8c45a7ace80e5bf05fffb704d9c91a0f022 Mon Sep 17 00:00:00 2001 From: kikiola Date: Tue, 28 Jul 2026 13:20:59 +0100 Subject: [PATCH] feat: implement notification system with BullMQ processor and user preference management --- docs/NOTIFICATION_ARCHITECTURE.md | 31 ++++++ src/app.module.ts | 4 + .../dto/create-notification.dto.ts | 36 +++++++ .../dto/update-preference.dto.ts | 43 ++++++++ .../notification-preference.entity.ts | 38 ++++++++ .../entities/notification.entity.ts | 43 ++++++++ .../enums/notification-category.enum.ts | 10 ++ .../enums/notification-channel.enum.ts | 6 ++ .../enums/notification-status.enum.ts | 7 ++ .../notifications.controller.spec.ts | 48 +++++++++ src/notifications/notifications.controller.ts | 73 ++++++++++++++ src/notifications/notifications.module.ts | 29 ++++++ .../notifications.processor.spec.ts | 67 +++++++++++++ src/notifications/notifications.processor.ts | 97 +++++++++++++++++++ .../notifications.service.spec.ts | 79 +++++++++++++++ src/notifications/notifications.service.ts | 97 +++++++++++++++++++ 16 files changed, 708 insertions(+) create mode 100644 docs/NOTIFICATION_ARCHITECTURE.md create mode 100644 src/notifications/dto/create-notification.dto.ts create mode 100644 src/notifications/dto/update-preference.dto.ts create mode 100644 src/notifications/entities/notification-preference.entity.ts create mode 100644 src/notifications/entities/notification.entity.ts create mode 100644 src/notifications/enums/notification-category.enum.ts create mode 100644 src/notifications/enums/notification-channel.enum.ts create mode 100644 src/notifications/enums/notification-status.enum.ts create mode 100644 src/notifications/notifications.controller.spec.ts create mode 100644 src/notifications/notifications.controller.ts create mode 100644 src/notifications/notifications.module.ts create mode 100644 src/notifications/notifications.processor.spec.ts create mode 100644 src/notifications/notifications.processor.ts create mode 100644 src/notifications/notifications.service.spec.ts create mode 100644 src/notifications/notifications.service.ts diff --git a/docs/NOTIFICATION_ARCHITECTURE.md b/docs/NOTIFICATION_ARCHITECTURE.md new file mode 100644 index 0000000..fd76e7c --- /dev/null +++ b/docs/NOTIFICATION_ARCHITECTURE.md @@ -0,0 +1,31 @@ +# Notification & Event Delivery Architecture + +## Overview +The Notification & Event Delivery API serves as the central communication backbone for the TruthBounty ecosystem. It handles real-time and asynchronous notifications across all platform modules (claims, verifications, disputes, governance, rewards). + +## Core Components + +### 1. Event Queue +Built on **BullMQ (Redis)**, ensuring fault-tolerant, asynchronous processing of events. +- **Queue Name**: `notifications` +- **Retry Mechanism**: Exponential backoff (max 5 attempts, delays starting at 2s, 4s, 8s, 16s). + +### 2. Preference Engine +Users define their notification preferences (`NotificationPreference` entity): +- **Channels**: `IN_APP`, `EMAIL`, `PUSH`, `WEBHOOK` +- **Categories**: `CLAIM`, `VERIFICATION`, `DISPUTE`, `GOVERNANCE`, `REWARD`, etc. +- **Quiet Hours & Digest Mode**: Settings stored for intelligent delivery windows. + +### 3. Delivery Tracking +The `Notification` entity tracks the lifecycle of every generated event: +- `QUEUED`, `DELIVERED`, `FAILED`, `READ`, `DISMISSED` +- Includes retry counts and detailed metadata for auditability. + +### 4. Metrics & Monitoring +Accessible via `GET /notifications/metrics` to expose: +- Delivery success rate +- Total processed notifications +- Queued / Failed events + +## Future Extensibility +The modular architecture permits drop-in integrations for SMS, Slack, Discord, Telegram, and AI-generated summaries without modifying the core queue processor. diff --git a/src/app.module.ts b/src/app.module.ts index 125f2d7..817899b 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -32,6 +32,9 @@ import { LoggingInterceptor } from './logger/logging.interceptor'; import { AuthModule } from './auth/auth.module'; import { GlobalAuthGuard } from './auth/global-auth.guard'; import { MetricsModule } from './metrics/metrics.module'; +import { NotificationsModule } from './notifications/notifications.module'; +import { Notification } from './notifications/entities/notification.entity'; +import { NotificationPreference } from './notifications/entities/notification-preference.entity'; // In-memory storage for development (no Redis needed) class ThrottlerMemoryStorage { @@ -283,6 +286,7 @@ async function createThrottlerStorage(configService: ConfigService): Promise; +} diff --git a/src/notifications/dto/update-preference.dto.ts b/src/notifications/dto/update-preference.dto.ts new file mode 100644 index 0000000..05ce5b6 --- /dev/null +++ b/src/notifications/dto/update-preference.dto.ts @@ -0,0 +1,43 @@ +import { IsArray, IsBoolean, IsEnum, IsOptional, IsString } from 'class-validator'; +import { NotificationChannel } from '../enums/notification-channel.enum'; +import { NotificationCategory } from '../enums/notification-category.enum'; +import { ApiPropertyOptional } from '@nestjs/swagger'; + +export class UpdatePreferenceDto { + @ApiPropertyOptional({ type: [String], enum: NotificationChannel }) + @IsArray() + @IsEnum(NotificationChannel, { each: true }) + @IsOptional() + enabledChannels?: string[]; + + @ApiPropertyOptional({ type: [String], enum: NotificationCategory }) + @IsArray() + @IsEnum(NotificationCategory, { each: true }) + @IsOptional() + disabledCategories?: string[]; + + @ApiPropertyOptional() + @IsBoolean() + @IsOptional() + digestMode?: boolean; + + @ApiPropertyOptional() + @IsBoolean() + @IsOptional() + quietHoursEnabled?: boolean; + + @ApiPropertyOptional() + @IsString() + @IsOptional() + quietHoursStart?: string; + + @ApiPropertyOptional() + @IsString() + @IsOptional() + quietHoursEnd?: string; + + @ApiPropertyOptional() + @IsString() + @IsOptional() + language?: string; +} diff --git a/src/notifications/entities/notification-preference.entity.ts b/src/notifications/entities/notification-preference.entity.ts new file mode 100644 index 0000000..aa8140a --- /dev/null +++ b/src/notifications/entities/notification-preference.entity.ts @@ -0,0 +1,38 @@ +import { Entity, Column, PrimaryGeneratedColumn, CreateDateColumn, UpdateDateColumn, Index } from 'typeorm'; + +@Entity('notification_preferences') +export class NotificationPreference { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ type: 'varchar', unique: true }) + @Index() + userId: string; + + @Column({ type: 'simple-json', default: '["IN_APP", "EMAIL"]' }) + enabledChannels: string[]; + + @Column({ type: 'simple-json', default: '[]' }) + disabledCategories: string[]; + + @Column({ type: 'boolean', default: false }) + digestMode: boolean; + + @Column({ type: 'boolean', default: false }) + quietHoursEnabled: boolean; + + @Column({ type: 'varchar', nullable: true }) + quietHoursStart: string; + + @Column({ type: 'varchar', nullable: true }) + quietHoursEnd: string; + + @Column({ type: 'varchar', default: 'en' }) + language: string; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/notifications/entities/notification.entity.ts b/src/notifications/entities/notification.entity.ts new file mode 100644 index 0000000..c227606 --- /dev/null +++ b/src/notifications/entities/notification.entity.ts @@ -0,0 +1,43 @@ +import { Entity, Column, PrimaryGeneratedColumn, CreateDateColumn, UpdateDateColumn } from 'typeorm'; +import { NotificationChannel } from '../enums/notification-channel.enum'; +import { NotificationCategory } from '../enums/notification-category.enum'; +import { NotificationStatus } from '../enums/notification-status.enum'; + +@Entity('notifications') +export class Notification { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ type: 'varchar' }) + userId: string; + + @Column({ type: 'varchar' }) + category: NotificationCategory; + + @Column({ type: 'varchar' }) + channel: NotificationChannel; + + @Column({ type: 'varchar' }) + title: string; + + @Column({ type: 'text' }) + content: string; + + @Column({ type: 'simple-json', nullable: true }) + metadata: Record; + + @Column({ type: 'varchar', default: NotificationStatus.QUEUED }) + status: NotificationStatus; + + @Column({ type: 'int', default: 0 }) + retryCount: number; + + @Column({ type: 'varchar', nullable: true }) + errorMessage: string; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/notifications/enums/notification-category.enum.ts b/src/notifications/enums/notification-category.enum.ts new file mode 100644 index 0000000..1c31c10 --- /dev/null +++ b/src/notifications/enums/notification-category.enum.ts @@ -0,0 +1,10 @@ +export enum NotificationCategory { + CLAIM = 'CLAIM', + VERIFICATION = 'VERIFICATION', + DISPUTE = 'DISPUTE', + GOVERNANCE = 'GOVERNANCE', + REWARD = 'REWARD', + REPUTATION = 'REPUTATION', + MODERATION = 'MODERATION', + SYSTEM = 'SYSTEM', +} diff --git a/src/notifications/enums/notification-channel.enum.ts b/src/notifications/enums/notification-channel.enum.ts new file mode 100644 index 0000000..aa7b8ec --- /dev/null +++ b/src/notifications/enums/notification-channel.enum.ts @@ -0,0 +1,6 @@ +export enum NotificationChannel { + IN_APP = 'IN_APP', + EMAIL = 'EMAIL', + PUSH = 'PUSH', + WEBHOOK = 'WEBHOOK', +} diff --git a/src/notifications/enums/notification-status.enum.ts b/src/notifications/enums/notification-status.enum.ts new file mode 100644 index 0000000..5173105 --- /dev/null +++ b/src/notifications/enums/notification-status.enum.ts @@ -0,0 +1,7 @@ +export enum NotificationStatus { + QUEUED = 'QUEUED', + DELIVERED = 'DELIVERED', + FAILED = 'FAILED', + READ = 'READ', + DISMISSED = 'DISMISSED', +} diff --git a/src/notifications/notifications.controller.spec.ts b/src/notifications/notifications.controller.spec.ts new file mode 100644 index 0000000..8a0ad5b --- /dev/null +++ b/src/notifications/notifications.controller.spec.ts @@ -0,0 +1,48 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { NotificationsController } from './notifications.controller'; +import { NotificationsService } from './notifications.service'; +import { NotificationCategory } from './enums/notification-category.enum'; + +describe('NotificationsController', () => { + let controller: NotificationsController; + let mockService: any; + + beforeEach(async () => { + mockService = { + queueNotification: jest.fn().mockResolvedValue({ id: '1' }), + getUserPreferences: jest.fn().mockResolvedValue({ userId: 'user-1' }), + updateUserPreferences: jest.fn().mockResolvedValue({ userId: 'user-1' }), + getDeliveryHistory: jest.fn().mockResolvedValue([[], 0]), + getMetrics: jest.fn().mockResolvedValue({ total: 0 }), + markAsRead: jest.fn().mockResolvedValue({ id: '1' }), + dismiss: jest.fn().mockResolvedValue({ id: '1' }), + }; + + const module: TestingModule = await Test.createTestingModule({ + controllers: [NotificationsController], + providers: [ + { + provide: NotificationsService, + useValue: mockService, + }, + ], + }).compile(); + + controller = module.get(NotificationsController); + }); + + it('should call queueNotification', async () => { + await controller.queueEvent({ + userId: '1', + category: NotificationCategory.CLAIM, + title: 'Test', + content: 'Test content', + }); + expect(mockService.queueNotification).toHaveBeenCalled(); + }); + + it('should call getPreferences', async () => { + await controller.getPreferences({ user: { id: 'user-1' } }); + expect(mockService.getUserPreferences).toHaveBeenCalledWith('user-1'); + }); +}); diff --git a/src/notifications/notifications.controller.ts b/src/notifications/notifications.controller.ts new file mode 100644 index 0000000..b41f444 --- /dev/null +++ b/src/notifications/notifications.controller.ts @@ -0,0 +1,73 @@ +import { Controller, Post, Body, Get, Param, Patch, Query, UseGuards, Request } from '@nestjs/common'; +import { NotificationsService } from './notifications.service'; +import { CreateNotificationDto } from './dto/create-notification.dto'; +import { UpdatePreferenceDto } from './dto/update-preference.dto'; +import { ApiTags, ApiOperation, ApiResponse, ApiBearerAuth, ApiQuery } from '@nestjs/swagger'; +import { GlobalAuthGuard } from '../auth/global-auth.guard'; + +@ApiTags('notifications') +@ApiBearerAuth() +@UseGuards(GlobalAuthGuard) +@Controller('notifications') +export class NotificationsController { + constructor(private readonly notificationsService: NotificationsService) {} + + @Post('event') + @ApiOperation({ summary: 'Queue a new protocol event notification' }) + @ApiResponse({ status: 201, description: 'The notification has been queued.' }) + async queueEvent(@Body() createDto: CreateNotificationDto) { + return this.notificationsService.queueNotification(createDto); + } + + @Get('preferences') + @ApiOperation({ summary: 'Get current user notification preferences' }) + async getPreferences(@Request() req) { + const userId = req.user?.id || req.user?.walletAddress || 'anonymous'; + return this.notificationsService.getUserPreferences(userId); + } + + @Patch('preferences') + @ApiOperation({ summary: 'Update user notification preferences' }) + async updatePreferences(@Request() req, @Body() updateDto: UpdatePreferenceDto) { + const userId = req.user?.id || req.user?.walletAddress || 'anonymous'; + return this.notificationsService.updateUserPreferences(userId, updateDto); + } + + @Get('history') + @ApiOperation({ summary: 'Get notification delivery history' }) + @ApiQuery({ name: 'skip', required: false, type: Number }) + @ApiQuery({ name: 'take', required: false, type: Number }) + async getHistory( + @Request() req, + @Query('skip') skip?: number, + @Query('take') take?: number, + ) { + const userId = req.user?.id || req.user?.walletAddress || 'anonymous'; + const [data, total] = await this.notificationsService.getDeliveryHistory( + userId, + skip ? Number(skip) : 0, + take ? Number(take) : 50, + ); + return { data, total, skip: skip || 0, take: take || 50 }; + } + + @Get('metrics') + @ApiOperation({ summary: 'Get notification delivery metrics' }) + async getMetrics() { + return this.notificationsService.getMetrics(); + } + + @Patch(':id/read') + @ApiOperation({ summary: 'Mark a notification as read' }) + async markAsRead(@Request() req, @Param('id') id: string) { + const userId = req.user?.id || req.user?.walletAddress || 'anonymous'; + return this.notificationsService.markAsRead(id, userId); + } + + @Patch(':id/dismiss') + @ApiOperation({ summary: 'Dismiss a notification' }) + async dismiss(@Request() req, @Param('id') id: string) { + const userId = req.user?.id || req.user?.walletAddress || 'anonymous'; + return this.notificationsService.dismiss(id, userId); + } +} diff --git a/src/notifications/notifications.module.ts b/src/notifications/notifications.module.ts new file mode 100644 index 0000000..aa59133 --- /dev/null +++ b/src/notifications/notifications.module.ts @@ -0,0 +1,29 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { BullModule } from '@nestjs/bullmq'; +import { BullBoardModule } from '@bull-board/nestjs'; +import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'; +import { NotificationsController } from './notifications.controller'; +import { NotificationsService } from './notifications.service'; +import { NotificationsProcessor } from './notifications.processor'; +import { Notification } from './entities/notification.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { AuthModule } from '../auth/auth.module'; + +@Module({ + imports: [ + TypeOrmModule.forFeature([Notification, NotificationPreference]), + BullModule.registerQueue({ + name: 'notifications', + }), + BullBoardModule.forFeature({ + name: 'notifications', + adapter: BullMQAdapter, + }), + AuthModule, + ], + controllers: [NotificationsController], + providers: [NotificationsService, NotificationsProcessor], + exports: [NotificationsService], +}) +export class NotificationsModule {} diff --git a/src/notifications/notifications.processor.spec.ts b/src/notifications/notifications.processor.spec.ts new file mode 100644 index 0000000..883f928 --- /dev/null +++ b/src/notifications/notifications.processor.spec.ts @@ -0,0 +1,67 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { NotificationsProcessor } from './notifications.processor'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { Notification } from './entities/notification.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { NotificationStatus } from './enums/notification-status.enum'; + +describe('NotificationsProcessor', () => { + let processor: NotificationsProcessor; + let mockNotificationRepository: any; + let mockPreferenceRepository: any; + + beforeEach(async () => { + mockNotificationRepository = { + findOne: jest.fn(), + save: jest.fn(), + }; + + mockPreferenceRepository = { + findOne: jest.fn(), + create: jest.fn().mockImplementation((dto) => dto), + save: jest.fn().mockImplementation((entity) => Promise.resolve(entity)), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + NotificationsProcessor, + { + provide: getRepositoryToken(Notification), + useValue: mockNotificationRepository, + }, + { + provide: getRepositoryToken(NotificationPreference), + useValue: mockPreferenceRepository, + }, + ], + }).compile(); + + processor = module.get(NotificationsProcessor); + }); + + it('should process a notification and mark as DELIVERED', async () => { + const notification = { id: '1', userId: 'user-1', status: NotificationStatus.QUEUED, category: 'SYSTEM' }; + mockNotificationRepository.findOne.mockResolvedValue(notification); + mockPreferenceRepository.findOne.mockResolvedValue({ userId: 'user-1', enabledChannels: ['IN_APP'] }); + + const job = { data: { notificationId: '1' }, attemptsMade: 0, opts: { attempts: 3 } } as any; + + await processor.process(job); + + expect(notification.status).toBe(NotificationStatus.DELIVERED); + expect(mockNotificationRepository.save).toHaveBeenCalledWith(notification); + }); + + it('should skip notification if category is disabled', async () => { + const notification = { id: '1', userId: 'user-1', status: NotificationStatus.QUEUED, category: 'SYSTEM' }; + mockNotificationRepository.findOne.mockResolvedValue(notification); + mockPreferenceRepository.findOne.mockResolvedValue({ userId: 'user-1', disabledCategories: ['SYSTEM'] }); + + const job = { data: { notificationId: '1' } } as any; + + await processor.process(job); + + expect(notification.status).toBe(NotificationStatus.DISMISSED); + expect(mockNotificationRepository.save).toHaveBeenCalledWith(notification); + }); +}); diff --git a/src/notifications/notifications.processor.ts b/src/notifications/notifications.processor.ts new file mode 100644 index 0000000..fbe5716 --- /dev/null +++ b/src/notifications/notifications.processor.ts @@ -0,0 +1,97 @@ +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Job } from 'bullmq'; +import { Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { Notification } from './entities/notification.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { NotificationStatus } from './enums/notification-status.enum'; +import { NotificationChannel } from './enums/notification-channel.enum'; + +@Processor('notifications') +export class NotificationsProcessor extends WorkerHost { + private readonly logger = new Logger(NotificationsProcessor.name); + + constructor( + @InjectRepository(Notification) + private notificationRepository: Repository, + @InjectRepository(NotificationPreference) + private preferenceRepository: Repository, + ) { + super(); + } + + async process(job: Job): Promise { + const { notificationId } = job.data; + this.logger.debug(`Processing notification ${notificationId}`); + + const notification = await this.notificationRepository.findOne({ where: { id: notificationId } }); + if (!notification) { + this.logger.error(`Notification ${notificationId} not found`); + return; + } + + try { + let pref = await this.preferenceRepository.findOne({ where: { userId: notification.userId } }); + if (!pref) { + pref = this.preferenceRepository.create({ userId: notification.userId }); + await this.preferenceRepository.save(pref); + } + + // Check category preferences + if (pref.disabledCategories && pref.disabledCategories.includes(notification.category)) { + this.logger.debug(`Notification ${notificationId} skipped due to disabled category ${notification.category}`); + notification.status = NotificationStatus.DISMISSED; + await this.notificationRepository.save(notification); + return; + } + + // Determine delivery channel + let channelToUse = notification.channel; + if (!channelToUse) { + // Fallback to IN_APP if no channel is explicitly provided, or use user's preferred channels + channelToUse = (pref.enabledChannels && pref.enabledChannels.length > 0) ? pref.enabledChannels[0] as NotificationChannel : NotificationChannel.IN_APP; + } else { + if (pref.enabledChannels && !pref.enabledChannels.includes(channelToUse)) { + this.logger.debug(`Notification ${notificationId} skipped due to disabled channel ${channelToUse}`); + notification.status = NotificationStatus.DISMISSED; + await this.notificationRepository.save(notification); + return; + } + } + + // Update notification with channel if it was missing + notification.channel = channelToUse; + + // Simulate quiet hours logic (simplified) + if (pref.quietHoursEnabled && pref.quietHoursStart && pref.quietHoursEnd) { + // In a real implementation, we would compare current time with quiet hours + // and optionally throw a delay error to queue it for later. + // this.logger.debug(`Quiet hours check for user ${notification.userId}`); + } + + // Simulate delivery based on channel + this.logger.log(`Delivering notification ${notificationId} via ${channelToUse}`); + + // Simulate failure randomly for testing retry mechanism (e.g. 10% chance) + // if (Math.random() < 0.1) { + // throw new Error('Simulated network failure'); + // } + + notification.status = NotificationStatus.DELIVERED; + await this.notificationRepository.save(notification); + this.logger.log(`Successfully delivered notification ${notificationId}`); + } catch (error) { + this.logger.error(`Failed to deliver notification ${notificationId}: ${error.message}`); + notification.retryCount += 1; + notification.errorMessage = error.message; + + if (job.attemptsMade >= (job.opts.attempts || 1)) { + notification.status = NotificationStatus.FAILED; + } + + await this.notificationRepository.save(notification); + throw error; // Let BullMQ handle retry + } + } +} diff --git a/src/notifications/notifications.service.spec.ts b/src/notifications/notifications.service.spec.ts new file mode 100644 index 0000000..32cb8d6 --- /dev/null +++ b/src/notifications/notifications.service.spec.ts @@ -0,0 +1,79 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { NotificationsService } from './notifications.service'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { getQueueToken } from '@nestjs/bullmq'; +import { Notification } from './entities/notification.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { NotificationStatus } from './enums/notification-status.enum'; +import { NotificationCategory } from './enums/notification-category.enum'; +import { NotificationChannel } from './enums/notification-channel.enum'; + +describe('NotificationsService', () => { + let service: NotificationsService; + let mockNotificationRepository: any; + let mockPreferenceRepository: any; + let mockQueue: any; + + beforeEach(async () => { + mockNotificationRepository = { + create: jest.fn().mockImplementation((dto) => dto), + save: jest.fn().mockImplementation((entity) => Promise.resolve({ id: '1', ...entity })), + findOne: jest.fn(), + findAndCount: jest.fn(), + count: jest.fn(), + }; + + mockPreferenceRepository = { + create: jest.fn().mockImplementation((dto) => dto), + save: jest.fn().mockImplementation((entity) => Promise.resolve({ id: '1', ...entity })), + findOne: jest.fn(), + }; + + mockQueue = { + add: jest.fn().mockResolvedValue(true), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + NotificationsService, + { + provide: getRepositoryToken(Notification), + useValue: mockNotificationRepository, + }, + { + provide: getRepositoryToken(NotificationPreference), + useValue: mockPreferenceRepository, + }, + { + provide: getQueueToken('notifications'), + useValue: mockQueue, + }, + ], + }).compile(); + + service = module.get(NotificationsService); + }); + + it('should queue a notification', async () => { + const dto = { + userId: 'user-1', + category: NotificationCategory.CLAIM, + title: 'New Claim', + content: 'A claim has been made.', + }; + + const result = await service.queueNotification(dto); + + expect(result.userId).toEqual('user-1'); + expect(result.status).toEqual(NotificationStatus.QUEUED); + expect(mockNotificationRepository.save).toHaveBeenCalled(); + expect(mockQueue.add).toHaveBeenCalledWith('send', { notificationId: '1' }, expect.any(Object)); + }); + + it('should return default preferences if not found', async () => { + mockPreferenceRepository.findOne.mockResolvedValue(null); + const result = await service.getUserPreferences('user-1'); + expect(result.userId).toEqual('user-1'); + expect(mockPreferenceRepository.save).toHaveBeenCalled(); + }); +}); diff --git a/src/notifications/notifications.service.ts b/src/notifications/notifications.service.ts new file mode 100644 index 0000000..e245474 --- /dev/null +++ b/src/notifications/notifications.service.ts @@ -0,0 +1,97 @@ +import { Injectable, Logger, NotFoundException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; +import { Notification } from './entities/notification.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { CreateNotificationDto } from './dto/create-notification.dto'; +import { UpdatePreferenceDto } from './dto/update-preference.dto'; +import { NotificationStatus } from './enums/notification-status.enum'; + +@Injectable() +export class NotificationsService { + private readonly logger = new Logger(NotificationsService.name); + + constructor( + @InjectRepository(Notification) + private notificationRepository: Repository, + @InjectRepository(NotificationPreference) + private preferenceRepository: Repository, + @InjectQueue('notifications') private notificationQueue: Queue, + ) {} + + async queueNotification(createDto: CreateNotificationDto): Promise { + const notification = this.notificationRepository.create(createDto); + notification.status = NotificationStatus.QUEUED; + const savedNotification = await this.notificationRepository.save(notification); + + await this.notificationQueue.add('send', { notificationId: savedNotification.id }, { + attempts: 5, + backoff: { + type: 'exponential', + delay: 2000, // 2s, 4s, 8s, 16s + }, + }); + + this.logger.log(`Queued notification ${savedNotification.id} for user ${savedNotification.userId}`); + return savedNotification; + } + + async getUserPreferences(userId: string): Promise { + let pref = await this.preferenceRepository.findOne({ where: { userId } }); + if (!pref) { + pref = this.preferenceRepository.create({ userId }); + pref = await this.preferenceRepository.save(pref); + } + return pref; + } + + async updateUserPreferences(userId: string, updateDto: UpdatePreferenceDto): Promise { + const pref = await this.getUserPreferences(userId); + Object.assign(pref, updateDto); + return this.preferenceRepository.save(pref); + } + + async getDeliveryHistory(userId: string, skip = 0, take = 50): Promise<[Notification[], number]> { + return this.notificationRepository.findAndCount({ + where: { userId }, + order: { createdAt: 'DESC' }, + skip, + take, + }); + } + + async getMetrics(): Promise { + const total = await this.notificationRepository.count(); + const delivered = await this.notificationRepository.count({ where: { status: NotificationStatus.DELIVERED } }); + const failed = await this.notificationRepository.count({ where: { status: NotificationStatus.FAILED } }); + const queued = await this.notificationRepository.count({ where: { status: NotificationStatus.QUEUED } }); + + return { + total, + delivered, + failed, + queued, + successRate: total > 0 ? (delivered / total) * 100 : 0, + }; + } + + async markAsRead(notificationId: string, userId: string): Promise { + const notification = await this.notificationRepository.findOne({ where: { id: notificationId, userId } }); + if (!notification) { + throw new NotFoundException('Notification not found'); + } + notification.status = NotificationStatus.READ; + return this.notificationRepository.save(notification); + } + + async dismiss(notificationId: string, userId: string): Promise { + const notification = await this.notificationRepository.findOne({ where: { id: notificationId, userId } }); + if (!notification) { + throw new NotFoundException('Notification not found'); + } + notification.status = NotificationStatus.DISMISSED; + return this.notificationRepository.save(notification); + } +}