diff --git a/src/notifications/controllers/index.ts b/src/notifications/controllers/index.ts new file mode 100644 index 0000000..05cf02f --- /dev/null +++ b/src/notifications/controllers/index.ts @@ -0,0 +1 @@ +export { NotificationsController } from './notifications.controller'; \ No newline at end of file diff --git a/src/notifications/controllers/notifications.controller.ts b/src/notifications/controllers/notifications.controller.ts new file mode 100644 index 0000000..0650188 --- /dev/null +++ b/src/notifications/controllers/notifications.controller.ts @@ -0,0 +1,88 @@ +import { + Controller, + Get, + Put, + Delete, + Param, + Body, + Query, + UseGuards, + HttpStatus, + HttpCode, +} from '@nestjs/common'; +import { JwtAuthGuard } from '../../auth/jwt-auth.guard'; +import { CurrentUser } from '../../common/decorators/current-user.decorator'; +import { NotificationsService } from '../services/notifications.service'; +import { NotificationPreferencesService } from '../services/notification-preferences.service'; +import { ListNotificationsDto, UpdatePreferencesDto } from '../dto'; +import { Notification } from '../entities/notification.entity'; + +@Controller('notifications') +@UseGuards(JwtAuthGuard) +export class NotificationsController { + constructor( + private readonly notificationsService: NotificationsService, + private readonly preferencesService: NotificationPreferencesService, + ) {} + + @Get() + async listNotifications( + @CurrentUser('id') userId: string, + @Query() filters: ListNotificationsDto, + ) { + return this.notificationsService.listNotifications(userId, filters); + } + + @Get('unread') + async getUnreadCount(@CurrentUser('id') userId: string) { + const count = await this.notificationsService.getUnreadCount(userId); + return { unreadCount: count }; + } + + @Get('history') + async getDeliveryHistory( + @CurrentUser('id') userId: string, + @Query() filters: ListNotificationsDto, + ) { + return this.notificationsService.getDeliveryHistory(userId, filters); + } + + @Put(':id/read') + @HttpCode(HttpStatus.OK) + async markAsRead( + @CurrentUser('id') userId: string, + @Param('id') notificationId: string, + ) { + await this.notificationsService.markAsRead(userId, notificationId); + return { success: true }; + } + + @Put('mark-all-read') + @HttpCode(HttpStatus.OK) + async markAllAsRead(@CurrentUser('id') userId: string) { + await this.notificationsService.markAllAsRead(userId); + return { success: true }; + } + + @Delete(':id') + @HttpCode(HttpStatus.NO_CONTENT) + async deleteNotification( + @CurrentUser('id') userId: string, + @Param('id') notificationId: string, + ) { + await this.notificationsService.deleteNotification(userId, notificationId); + } + + @Get('preferences') + async getPreferences(@CurrentUser('id') userId: string) { + return this.preferencesService.getUserPreferences(userId); + } + + @Put('preferences') + async updatePreferences( + @CurrentUser('id') userId: string, + @Body() preferences: UpdatePreferencesDto, + ) { + return this.preferencesService.updateUserPreferences(userId, preferences); + } +} \ No newline at end of file diff --git a/src/notifications/dto/index.ts b/src/notifications/dto/index.ts new file mode 100644 index 0000000..d2a9264 --- /dev/null +++ b/src/notifications/dto/index.ts @@ -0,0 +1,2 @@ +export { ListNotificationsDto } from './list-notifications.dto'; +export { UpdatePreferencesDto } from './update-preferences.dto'; \ No newline at end of file diff --git a/src/notifications/dto/list-notifications.dto.ts b/src/notifications/dto/list-notifications.dto.ts new file mode 100644 index 0000000..df73138 --- /dev/null +++ b/src/notifications/dto/list-notifications.dto.ts @@ -0,0 +1,39 @@ +import { IsOptional, IsString, IsBoolean, IsInt, Min, Max, IsEnum } from 'class-validator'; +import { Type } from 'class-transformer'; +import { NotificationCategory, NotificationPriority } from '../interfaces/notification.types'; + +export class ListNotificationsDto { + @IsOptional() + @IsInt() + @Min(1) + @Type(() => Number) + page?: number = 1; + + @IsOptional() + @IsInt() + @Min(1) + @Max(100) + @Type(() => Number) + limit?: number = 20; + + @IsOptional() + @IsBoolean() + @Type(() => Boolean) + unreadOnly?: boolean; + + @IsOptional() + @IsEnum(NotificationCategory) + category?: NotificationCategory; + + @IsOptional() + @IsEnum(NotificationPriority) + priority?: NotificationPriority; + + @IsOptional() + @IsString() + fromDate?: string; + + @IsOptional() + @IsString() + toDate?: string; +} \ No newline at end of file diff --git a/src/notifications/dto/update-preferences.dto.ts b/src/notifications/dto/update-preferences.dto.ts new file mode 100644 index 0000000..188a923 --- /dev/null +++ b/src/notifications/dto/update-preferences.dto.ts @@ -0,0 +1,48 @@ +import { IsOptional, IsBoolean, IsArray, IsEnum, ValidateNested } from 'class-validator'; +import { Type } from 'class-transformer'; +import { DeliveryChannel, NotificationCategory } from '../interfaces/notification.types'; + +export class EmailPreferencesDto { + @IsOptional() + @IsBoolean() + digestEnabled?: boolean; + + @IsOptional() + @IsEnum(['daily', 'weekly', 'never']) + digestFrequency?: 'daily' | 'weekly' | 'never'; + + @IsOptional() + @IsString() + emailAddress?: string; +} + +export class UpdatePreferencesDto { + @IsOptional() + @IsArray() + @IsEnum(DeliveryChannel, { each: true }) + enabledChannels?: DeliveryChannel[]; + + @IsOptional() + @ValidateNested() + @Type(() => EmailPreferencesDto) + emailPreferences?: EmailPreferencesDto; + + @IsOptional() + @IsBoolean() + governanceAlerts?: boolean; + + @IsOptional() + @IsBoolean() + stakingAlerts?: boolean; + + @IsOptional() + @IsBoolean() + rewardNotifications?: boolean; + + @IsOptional() + @IsBoolean() + securityAlerts?: boolean; + + @IsOptional() + categorySettings?: Record; +} \ No newline at end of file diff --git a/src/notifications/entities/delivery-history.entity.ts b/src/notifications/entities/delivery-history.entity.ts new file mode 100644 index 0000000..6e2d49e --- /dev/null +++ b/src/notifications/entities/delivery-history.entity.ts @@ -0,0 +1,46 @@ +import { Entity, Column, PrimaryGeneratedColumn, ManyToOne, CreateDateColumn } from 'typeorm'; +import { Notification } from './notification.entity'; +import { DeliveryChannel, DeliveryStatus } from '../interfaces/notification.types'; + +@Entity('delivery_history') +export class DeliveryHistory { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + notificationId: string; + + @ManyToOne(() => Notification, { onDelete: 'CASCADE' }) + notification: Notification; + + @Column({ + type: 'varchar', + length: 20, + }) + channel: DeliveryChannel; + + @Column({ + type: 'varchar', + length: 20, + default: 'pending', + }) + status: DeliveryStatus; + + @Column({ type: 'int', default: 0 }) + retryAttempts: number; + + @Column({ nullable: true }) + lastRetryAt: Date; + + @Column({ nullable: true }) + deliveredAt: Date; + + @Column({ nullable: true }) + failureReason: string; + + @Column({ type: 'jsonb', nullable: true }) + metadata: Record; + + @CreateDateColumn() + createdAt: Date; +} \ No newline at end of file diff --git a/src/notifications/entities/index.ts b/src/notifications/entities/index.ts new file mode 100644 index 0000000..e6d4cdc --- /dev/null +++ b/src/notifications/entities/index.ts @@ -0,0 +1,3 @@ +export { Notification } from './notification.entity'; +export { NotificationPreference } from './notification-preference.entity'; +export { DeliveryHistory } from './delivery-history.entity'; \ No newline at end of file diff --git a/src/notifications/entities/notification-preference.entity.ts b/src/notifications/entities/notification-preference.entity.ts new file mode 100644 index 0000000..a350ff2 --- /dev/null +++ b/src/notifications/entities/notification-preference.entity.ts @@ -0,0 +1,25 @@ +import { Entity, Column, PrimaryGeneratedColumn, OneToOne, JoinColumn } from 'typeorm'; +import { User } from '../../entities/user.entity'; +import { UserPreferenceSettings } from '../interfaces/notification.types'; + +@Entity('notification_preferences') +export class NotificationPreference { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ unique: true }) + userId: string; + + @OneToOne(() => User, { onDelete: 'CASCADE' }) + @JoinColumn() + user: User; + + @Column({ type: 'jsonb' }) + settings: UserPreferenceSettings; + + @Column({ default: () => 'now()' }) + createdAt: Date; + + @Column({ default: () => 'now()' }) + updatedAt: Date; +} \ No newline at end of file diff --git a/src/notifications/entities/notification.entity.ts b/src/notifications/entities/notification.entity.ts index 71532be..3d94325 100644 --- a/src/notifications/entities/notification.entity.ts +++ b/src/notifications/entities/notification.entity.ts @@ -1,3 +1,8 @@ +import { Entity, Column, PrimaryGeneratedColumn, ManyToOne, CreateDateColumn, UpdateDateColumn } from 'typeorm'; +import { User } from '../../entities/user.entity'; +import { NotificationCategory, NotificationPriority } from '../interfaces/notification.types'; + +@Entity('notifications') import { Entity, PrimaryGeneratedColumn, @@ -30,6 +35,41 @@ export class Notification { userId: string; @ManyToOne(() => User, { onDelete: 'CASCADE' }) + user: User; + + @Column() + title: string; + + @Column('text') + message: string; + + @Column({ + type: 'varchar', + length: 50, + }) + category: NotificationCategory; + + @Column({ + type: 'varchar', + length: 20, + default: 'medium', + }) + priority: NotificationPriority; + + @Column({ type: 'jsonb', nullable: true }) + metadata: Record; + + @Column({ type: 'jsonb', nullable: true }) + sourceEvent: Record; + + @Column({ default: false }) + read: boolean; + + @Column({ nullable: true }) + readAt: Date; + + @Column({ nullable: true }) + scheduledFor: Date; @JoinColumn({ name: 'userId' }) user: User; @@ -75,3 +115,4 @@ export class Notification { @UpdateDateColumn() updatedAt: Date; } +} diff --git a/src/notifications/interfaces/notification.types.ts b/src/notifications/interfaces/notification.types.ts new file mode 100644 index 0000000..b77d78e --- /dev/null +++ b/src/notifications/interfaces/notification.types.ts @@ -0,0 +1,90 @@ +export enum NotificationCategory { + NEW_CLAIM = 'new_claim', + VERIFICATION_ASSIGNMENT = 'verification_assignment', + DISPUTE_CREATED = 'dispute_created', + DISPUTE_RESOLVED = 'dispute_resolved', + REPUTATION_UPDATE = 'reputation_update', + STAKING_CHANGE = 'staking_change', + GOVERNANCE_PROPOSAL = 'governance_proposal', + PROPOSAL_VOTING = 'proposal_voting', + REWARD_DISTRIBUTION = 'reward_distribution', + MODERATION_ACTION = 'moderation_action', + SECURITY_ALERT = 'security_alert', + VOTING_DEADLINE = 'voting_deadline', + SYSTEM_UPDATE = 'system_update', +} + +export enum NotificationPriority { + LOW = 'low', + MEDIUM = 'medium', + HIGH = 'high', + CRITICAL = 'critical', +} + +export enum DeliveryChannel { + IN_APP = 'in_app', + WEBSOCKET = 'websocket', + PUSH = 'push', + EMAIL = 'email', + WEBHOOK = 'webhook', +} + +export enum DeliveryStatus { + PENDING = 'pending', + DELIVERED = 'delivered', + FAILED = 'failed', + PERMANENT_FAILURE = 'permanent_failure', + RETRYING = 'retrying', +} + +export interface NotificationEvent { + eventType: string; + source: string; + timestamp: Date; + payload: Record; + recipientIds: string[]; +} + +export interface DeliveryResult { + success: boolean; + status: DeliveryStatus; + error?: string; + deliveredAt?: Date; +} + +export interface UserPreferenceSettings { + enabledChannels: DeliveryChannel[]; + categories: { + [key in NotificationCategory]?: boolean; + }; + emailPreferences: { + digestEnabled: boolean; + digestFrequency: 'daily' | 'weekly' | 'never'; + emailAddress?: string; + }; + governanceAlerts: boolean; + stakingAlerts: boolean; + rewardNotifications: boolean; + securityAlerts: boolean; +} + +export interface WebhookConfig { + id: string; + url: string; + secret: string; + events: string[]; + userId: string; + active: boolean; + createdAt: Date; + updatedAt: Date; +} + +export interface PushSubscription { + id: string; + endpoint: string; + p256dh: string; + auth: string; + userId: string; + userAgent?: string; + createdAt: Date; +} \ No newline at end of file diff --git a/src/notifications/notifications.module.ts b/src/notifications/notifications.module.ts new file mode 100644 index 0000000..fa6062e --- /dev/null +++ b/src/notifications/notifications.module.ts @@ -0,0 +1,53 @@ +import { Module } from '@nestjs/common'; +import { BullModule } from '@nestjs/bullmq'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { PrismaModule } from '../prisma/prisma.module'; +import { RedisModule } from '../redis/redis.module'; +import { LoggerModule } from '../logger/logger.module'; +import { NotificationsController } from './controllers/notifications.controller'; +import { NotificationsService } from './services/notifications.service'; +import { NotificationPreferencesService } from './services/notification-preferences.service'; +import { DeliveryHistoryService } from './services/delivery-history.service'; +import { WebSocketService } from './services/websocket.service'; +import { EmailService } from './services/email.service'; +import { WebhookService } from './services/webhook.service'; +import { NotificationProcessor } from './services/notification.processor'; +import { Notification } from './entities/notification.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { DeliveryHistory } from './entities/delivery-history.entity'; +import { MetricsModule } from '../metrics/metrics.module'; +import { AuthModule } from '../auth/auth.module'; + +@Module({ + imports: [ + TypeOrmModule.forFeature([Notification, NotificationPreference, DeliveryHistory]), + BullModule.registerQueue({ + name: 'notifications', + defaultAttempts: 5, + defaultBackoff: { + type: 'exponential', + delay: 1000, + }, + }), + BullModule.registerQueue({ + name: 'dead-letter', + }), + PrismaModule, + RedisModule, + LoggerModule, + MetricsModule, + AuthModule, + ], + controllers: [NotificationsController], + providers: [ + NotificationsService, + NotificationPreferencesService, + DeliveryHistoryService, + WebSocketService, + EmailService, + WebhookService, + NotificationProcessor, + ], + exports: [NotificationsService], +}) +export class NotificationsModule {} \ No newline at end of file diff --git a/src/notifications/services/delivery-history.service.ts b/src/notifications/services/delivery-history.service.ts new file mode 100644 index 0000000..b3bf7f6 --- /dev/null +++ b/src/notifications/services/delivery-history.service.ts @@ -0,0 +1,133 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { DeliveryHistory } from '../entities/delivery-history.entity'; +import { ListNotificationsDto } from '../dto'; +import { DeliveryStatus, DeliveryChannel } from '../interfaces/notification.types'; + +@Injectable() +export class DeliveryHistoryService { + private readonly logger = new Logger(DeliveryHistoryService.name); + + constructor( + @InjectRepository(DeliveryHistory) + private readonly deliveryHistoryRepository: Repository, + ) {} + + async createDeliveryRecord( + notificationId: string, + channel: DeliveryChannel + ): Promise { + const record = new DeliveryHistory(); + record.notificationId = notificationId; + record.channel = channel; + record.status = DeliveryStatus.PENDING; + record.retryAttempts = 0; + record.createdAt = new Date(); + + return this.deliveryHistoryRepository.save(record); + } + + async updateDeliveryStatus( + recordId: string, + status: DeliveryStatus, + error?: string + ): Promise { + const record = await this.deliveryHistoryRepository.findOne({ + where: { id: recordId }, + }); + + if (!record) { + this.logger.error(`Delivery record ${recordId} not found`); + return null; + } + + record.status = status; + + if (status === DeliveryStatus.DELIVERED) { + record.deliveredAt = new Date(); + } + + if (error) { + record.failureReason = error; + } + + return this.deliveryHistoryRepository.save(record); + } + + async incrementRetryAttempts(recordId: string): Promise { + const record = await this.deliveryHistoryRepository.findOne({ + where: { id: recordId }, + }); + + if (!record) { + this.logger.error(`Delivery record ${recordId} not found`); + return null; + } + + record.retryAttempts += 1; + record.lastRetryAt = new Date(); + record.status = DeliveryStatus.RETRYING; + + return this.deliveryHistoryRepository.save(record); + } + + async getUserDeliveryHistory(userId: string, filters: ListNotificationsDto) { + const { page = 1, limit = 20 } = filters; + + const queryBuilder = this.deliveryHistoryRepository.createQueryBuilder('history') + .leftJoinAndSelect('history.notification', 'notification') + .where('notification.userId = :userId', { userId }) + .orderBy('history.createdAt', 'DESC') + .skip((page - 1) * limit) + .take(limit); + + const [items, total] = await queryBuilder.getManyAndCount(); + + return { + items, + total, + page, + limit, + totalPages: Math.ceil(total / limit), + }; + } + + async getDeliveryMetrics() { + const totalDeliveries = await this.deliveryHistoryRepository.count(); + const successfulDeliveries = await this.deliveryHistoryRepository.count({ + where: { status: DeliveryStatus.DELIVERED }, + }); + const failedDeliveries = await this.deliveryHistoryRepository.count({ + where: { status: DeliveryStatus.FAILED }, + }); + const pendingDeliveries = await this.deliveryHistoryRepository.count({ + where: { status: DeliveryStatus.PENDING }, + }); + const retryingDeliveries = await this.deliveryHistoryRepository.count({ + where: { status: DeliveryStatus.RETRYING }, + }); + + return { + totalDeliveries, + successfulDeliveries, + failedDeliveries, + pendingDeliveries, + retryingDeliveries, + successRate: totalDeliveries > 0 ? (successfulDeliveries / totalDeliveries) * 100 : 0, + }; + } + + async findPendingDeliveryByNotificationAndChannel( + notificationId: string, + channel: DeliveryChannel + ): Promise { + return this.deliveryHistoryRepository.findOne({ + where: { + notificationId, + channel, + status: DeliveryStatus.PENDING, + }, + }); + } +} \ No newline at end of file diff --git a/src/notifications/services/email.service.ts b/src/notifications/services/email.service.ts new file mode 100644 index 0000000..eaf3c9a --- /dev/null +++ b/src/notifications/services/email.service.ts @@ -0,0 +1,171 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { Notification } from '../entities/notification.entity'; +import { DeliveryResult, DeliveryStatus } from '../interfaces/notification.types'; + +@Injectable() +export class EmailService { + private readonly logger = new Logger(EmailService.name); + private fromEmail: string; + private smtpConfigured: boolean = false; + + constructor(private readonly configService: ConfigService) { + this.initializeEmailConfig(); + } + + private initializeEmailConfig() { + this.fromEmail = this.configService.get('EMAIL_FROM', 'notifications@truthbounty.io'); + const smtpHost = this.configService.get('SMTP_HOST'); + const smtpPort = this.configService.get('SMTP_PORT'); + + this.smtpConfigured = !!(smtpHost && smtpPort); + + if (this.smtpConfigured) { + this.logger.log('Email service configured successfully'); + } else { + this.logger.warn('SMTP not configured, email notifications will be simulated'); + } + } + + async sendNotificationEmail( + notification: Notification, + recipientEmail: string + ): Promise { + this.logger.debug(`Preparing to send email to ${recipientEmail} for notification ${notification.id}`); + + if (!this.smtpConfigured) { + this.logger.info(`[SIMULATED] Email would be sent to ${recipientEmail}: ${notification.title}`); + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + } + + try { + const emailContent = this.generateEmailContent(notification); + + await this.sendEmail( + recipientEmail, + notification.title, + emailContent.html, + emailContent.text + ); + + this.logger.log(`Email sent successfully to ${recipientEmail}`); + + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + } catch (error) { + this.logger.error(`Failed to send email to ${recipientEmail}`, error); + return { + success: false, + status: DeliveryStatus.FAILED, + error: error.message, + }; + } + } + + private generateEmailContent(notification: Notification) { + const html = ` + + + + ${notification.title} + + + +
+

${notification.title}

+
+
+

${notification.message}

+ ${this.generateMetadataHtml(notification.metadata)} +
+
+

This email was sent by TruthBounty. You can manage your notification preferences in your account settings.

+
+ + + `; + + const text = ` +${notification.title} + +${notification.message} + +${this.generateMetadataText(notification.metadata)} + +--- +This email was sent by TruthBounty. You can manage your notification preferences in your account settings. + `; + + return { html, text }; + } + + private generateMetadataHtml(metadata?: Record): string { + if (!metadata || Object.keys(metadata).length === 0) return ''; + + let html = '
'; + html += '

Additional Details

'; + html += '
    '; + + for (const [key, value] of Object.entries(metadata)) { + html += `
  • ${key}: ${value}
    • `; + } + + html += '
'; + return html; + } + + private generateMetadataText(metadata?: Record): string { + if (!metadata || Object.keys(metadata).length === 0) return ''; + + let text = '\nAdditional Details:\n'; + for (const [key, value] of Object.entries(metadata)) { + text += `${key}: ${value}\n`; + } + return text; + } + + private async sendEmail( + to: string, + subject: string, + html: string, + text: string + ): Promise { + if (!this.smtpConfigured) return; + + this.logger.debug(`Actually sending email to ${to}: ${subject}`); + } + + async sendDigestEmail( + userId: string, + notifications: Notification[], + frequency: 'daily' | 'weekly' + ): Promise { + this.logger.log(`Sending ${frequency} digest to user ${userId} with ${notifications.length} notifications`); + + if (notifications.length === 0) { + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + } + + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + } +} \ No newline at end of file diff --git a/src/notifications/services/index.ts b/src/notifications/services/index.ts new file mode 100644 index 0000000..3096814 --- /dev/null +++ b/src/notifications/services/index.ts @@ -0,0 +1,7 @@ +export { NotificationsService } from './notifications.service'; +export { NotificationPreferencesService } from './notification-preferences.service'; +export { DeliveryHistoryService } from './delivery-history.service'; +export { WebSocketService } from './websocket.service'; +export { EmailService } from './email.service'; +export { WebhookService } from './webhook.service'; +export { NotificationProcessor } from './notification.processor'; \ No newline at end of file diff --git a/src/notifications/services/notification-preferences.service.ts b/src/notifications/services/notification-preferences.service.ts new file mode 100644 index 0000000..1358c06 --- /dev/null +++ b/src/notifications/services/notification-preferences.service.ts @@ -0,0 +1,112 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { NotificationPreference } from '../entities/notification-preference.entity'; +import { UpdatePreferencesDto } from '../dto'; +import { + UserPreferenceSettings, + DeliveryChannel, + NotificationCategory +} from '../interfaces/notification.types'; + +@Injectable() +export class NotificationPreferencesService { + private readonly logger = new Logger(NotificationPreferencesService.name); + + constructor( + @InjectRepository(NotificationPreference) + private readonly preferencesRepository: Repository, + ) {} + + async getUserPreferences(userId: string): Promise { + let preferences = await this.preferencesRepository.findOne({ + where: { userId }, + }); + + if (!preferences) { + preferences = this.createDefaultPreferences(userId); + await this.preferencesRepository.save(preferences); + this.logger.debug(`Created default preferences for user ${userId}`); + } + + return preferences; + } + + async updateUserPreferences( + userId: string, + updateDto: UpdatePreferencesDto + ): Promise { + const preferences = await this.getUserPreferences(userId); + + if (updateDto.enabledChannels) { + preferences.settings.enabledChannels = updateDto.enabledChannels; + } + + if (updateDto.emailPreferences) { + preferences.settings.emailPreferences = { + ...preferences.settings.emailPreferences, + ...updateDto.emailPreferences, + }; + } + + if (updateDto.governanceAlerts !== undefined) { + preferences.settings.governanceAlerts = updateDto.governanceAlerts; + } + + if (updateDto.stakingAlerts !== undefined) { + preferences.settings.stakingAlerts = updateDto.stakingAlerts; + } + + if (updateDto.rewardNotifications !== undefined) { + preferences.settings.rewardNotifications = updateDto.rewardNotifications; + } + + if (updateDto.securityAlerts !== undefined) { + preferences.settings.securityAlerts = updateDto.securityAlerts; + } + + if (updateDto.categorySettings) { + preferences.settings.categories = { + ...preferences.settings.categories, + ...updateDto.categorySettings, + }; + } + + preferences.updatedAt = new Date(); + const updatedPreferences = await this.preferencesRepository.save(preferences); + + this.logger.log(`Updated preferences for user ${userId}`); + return updatedPreferences; + } + + private createDefaultPreferences(userId: string): NotificationPreference { + const defaultSettings: UserPreferenceSettings = { + enabledChannels: [DeliveryChannel.IN_APP, DeliveryChannel.WEBSOCKET, DeliveryChannel.EMAIL], + categories: this.getDefaultCategorySettings(), + emailPreferences: { + digestEnabled: false, + digestFrequency: 'daily', + }, + governanceAlerts: true, + stakingAlerts: true, + rewardNotifications: true, + securityAlerts: true, + }; + + const preferences = new NotificationPreference(); + preferences.userId = userId; + preferences.settings = defaultSettings; + preferences.createdAt = new Date(); + preferences.updatedAt = new Date(); + + return preferences; + } + + private getDefaultCategorySettings(): Record { + const categories = Object.values(NotificationCategory); + return categories.reduce((acc, category) => { + acc[category] = true; + return acc; + }, {} as Record); + } +} \ No newline at end of file diff --git a/src/notifications/services/notification.processor.ts b/src/notifications/services/notification.processor.ts new file mode 100644 index 0000000..8a22cdc --- /dev/null +++ b/src/notifications/services/notification.processor.ts @@ -0,0 +1,156 @@ +import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq'; +import { Job } from 'bullmq'; +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { Notification } from '../entities/notification.entity'; +import { WebSocketService } from './websocket.service'; +import { EmailService } from './email.service'; +import { WebhookService } from './webhook.service'; +import { DeliveryHistoryService } from './delivery-history.service'; +import { + DeliveryChannel, + DeliveryStatus, + NotificationDeliveryJob +} from '../interfaces/notification.types'; + +@Processor('notifications', { + concurrency: 10, +}) +@Injectable() +export class NotificationProcessor extends WorkerHost { + private readonly logger = new Logger(NotificationProcessor.name); + + constructor( + @InjectRepository(Notification) + private readonly notificationRepository: Repository, + private readonly webSocketService: WebSocketService, + private readonly emailService: EmailService, + private readonly webhookService: WebhookService, + private readonly deliveryHistoryService: DeliveryHistoryService, + ) { + super(); + } + + async process(job: Job): Promise { + const { notificationId, channel } = job.data; + this.logger.debug(`Processing notification delivery: ${notificationId} via ${channel} (attempt ${job.attemptsMade + 1})`); + + const deliveryRecord = await this.deliveryHistoryService.findPendingDeliveryByNotificationAndChannel( + notificationId, + channel + ); + + if (!deliveryRecord) { + this.logger.warn(`No pending delivery record found for ${notificationId} via ${channel}`); + return; + } + + if (job.attemptsMade > 0) { + await this.deliveryHistoryService.incrementRetryAttempts(deliveryRecord.id); + } + + const notification = await this.notificationRepository.findOne({ + where: { id: notificationId }, + }); + + if (!notification) { + this.logger.error(`Notification ${notificationId} not found`); + await this.deliveryHistoryService.updateDeliveryStatus( + deliveryRecord.id, + DeliveryStatus.FAILED, + 'Notification not found in database' + ); + return; + } + + const result = await this.deliverViaChannel(notification, channel); + + await this.deliveryHistoryService.updateDeliveryStatus( + deliveryRecord.id, + result.status, + result.error + ); + + if (!result.success) { + throw new Error(result.error || 'Delivery failed'); + } + + this.logger.log(`Notification ${notificationId} delivered successfully via ${channel}`); + return result; + } + + private async deliverViaChannel(notification: Notification, channel: DeliveryChannel) { + switch (channel) { + case DeliveryChannel.WEBSOCKET: + return this.webSocketService.broadcastNotification(notification); + + case DeliveryChannel.EMAIL: + return this.emailService.sendNotificationEmail(notification, 'user@example.com'); + + case DeliveryChannel.WEBHOOK: + const userWebhooks = await this.webhookService.getUserWebhooks(notification.userId); + if (userWebhooks.length > 0) { + const webhookResults = await Promise.all( + userWebhooks.map(webhook => + this.webhookService.sendWebhookNotification(notification, webhook) + ) + ); + const allSuccessful = webhookResults.every(r => r.success); + return { + success: allSuccessful, + status: allSuccessful ? DeliveryStatus.DELIVERED : DeliveryStatus.FAILED, + deliveredAt: allSuccessful ? new Date() : undefined, + error: allSuccessful ? undefined : 'Some webhooks failed to deliver', + }; + } + return { + success: false, + status: DeliveryStatus.FAILED, + error: 'No webhooks configured for user', + }; + + case DeliveryChannel.IN_APP: + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + + case DeliveryChannel.PUSH: + this.logger.debug(`Push notifications not yet implemented`); + return { + success: false, + status: DeliveryStatus.FAILED, + error: 'Push notifications not implemented', + }; + + default: + return { + success: false, + status: DeliveryStatus.FAILED, + error: `Unknown delivery channel: ${channel}`, + }; + } + } + + @OnWorkerEvent('completed') + onCompleted(job: Job) { + this.logger.debug(`Job ${job.id} completed successfully`); + } + + @OnWorkerEvent('failed') + onFailed(job: Job, error: Error) { + this.logger.error(`Job ${job.id} failed after ${job.attemptsMade} attempts: ${error.message}`); + } + + @OnWorkerEvent('error') + onError(error: Error) { + this.logger.error('Worker encountered an error:', error); + } + + @OnWorkerEvent('stalled') + onStalled(jobId: string) { + this.logger.warn(`Job ${jobId} has stalled`); + } +} \ No newline at end of file diff --git a/src/notifications/services/notifications.service.ts b/src/notifications/services/notifications.service.ts new file mode 100644 index 0000000..a237489 --- /dev/null +++ b/src/notifications/services/notifications.service.ts @@ -0,0 +1,298 @@ +import { Injectable, Logger, NotFoundException, ForbiddenException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Repository, Between, LessThanOrEqual, MoreThanOrEqual } from 'typeorm'; +import { Queue } from 'bullmq'; +import { Notification } from '../entities/notification.entity'; +import { NotificationPreference } from '../entities/notification-preference.entity'; +import { DeliveryHistoryService } from './delivery-history.service'; +import { NotificationPreferencesService } from './notification-preferences.service'; +import { ListNotificationsDto } from '../dto'; +import { NotificationEvent, NotificationCategory, NotificationPriority } from '../interfaces/notification.types'; +import { MetricsService } from '../../metrics/metrics.service'; + +@Injectable() +export class NotificationsService { + private readonly logger = new Logger(NotificationsService.name); + + constructor( + @InjectRepository(Notification) + private readonly notificationRepository: Repository, + @InjectQueue('notifications') + private readonly notificationsQueue: Queue, + private readonly deliveryHistoryService: DeliveryHistoryService, + private readonly preferencesService: NotificationPreferencesService, + private readonly metricsService: MetricsService, + ) {} + + async listNotifications(userId: string, filters: ListNotificationsDto) { + const { page = 1, limit = 20, unreadOnly, category, priority, fromDate, toDate } = filters; + const queryBuilder = this.notificationRepository.createQueryBuilder('notification'); + + queryBuilder.where('notification.userId = :userId', { userId }); + + if (unreadOnly) { + queryBuilder.andWhere('notification.read = false'); + } + + if (category) { + queryBuilder.andWhere('notification.category = :category', { category }); + } + + if (priority) { + queryBuilder.andWhere('notification.priority = :priority', { priority }); + } + + if (fromDate && toDate) { + queryBuilder.andWhere('notification.createdAt BETWEEN :fromDate AND :toDate', { + fromDate: new Date(fromDate), + toDate: new Date(toDate), + }); + } else if (fromDate) { + queryBuilder.andWhere('notification.createdAt >= :fromDate', { fromDate: new Date(fromDate) }); + } else if (toDate) { + queryBuilder.andWhere('notification.createdAt <= :toDate', { toDate: new Date(toDate) }); + } + + queryBuilder.orderBy('notification.createdAt', 'DESC') + .skip((page - 1) * limit) + .take(limit); + + const [items, total] = await queryBuilder.getManyAndCount(); + + return { + items, + total, + page, + limit, + totalPages: Math.ceil(total / limit), + }; + } + + async getUnreadCount(userId: string): Promise { + return this.notificationRepository.count({ + where: { userId, read: false }, + }); + } + + async getDeliveryHistory(userId: string, filters: ListNotificationsDto) { + return this.deliveryHistoryService.getUserDeliveryHistory(userId, filters); + } + + async markAsRead(userId: string, notificationId: string): Promise { + const notification = await this.notificationRepository.findOne({ + where: { id: notificationId, userId }, + }); + + if (!notification) { + throw new NotFoundException('Notification not found'); + } + + if (notification.userId !== userId) { + throw new ForbiddenException('You do not have access to this notification'); + } + + notification.read = true; + notification.readAt = new Date(); + await this.notificationRepository.save(notification); + } + + async markAllAsRead(userId: string): Promise { + await this.notificationRepository.update( + { userId, read: false }, + { read: true, readAt: new Date() } + ); + } + + async deleteNotification(userId: string, notificationId: string): Promise { + const notification = await this.notificationRepository.findOne({ + where: { id: notificationId, userId }, + }); + + if (!notification) { + throw new NotFoundException('Notification not found'); + } + + if (notification.userId !== userId) { + throw new ForbiddenException('You do not have access to this notification'); + } + + await this.notificationRepository.remove(notification); + } + + async processIncomingEvent(event: NotificationEvent): Promise { + this.logger.log(`Processing incoming event: ${event.eventType} from ${event.source}`); + + for (const recipientId of event.recipientIds) { + try { + const preferences = await this.preferencesService.getUserPreferences(recipientId); + + if (!this.shouldSendNotification(preferences, event)) { + this.logger.debug(`User ${recipientId} has disabled notifications for ${event.eventType}`); + continue; + } + + const notification = this.createNotificationFromEvent(event, recipientId); + const savedNotification = await this.notificationRepository.save(notification); + + await this.queueNotificationForDelivery(savedNotification, preferences); + + this.metricsService.incrementCounter('notifications_created_total', 1); + this.logger.debug(`Notification created for user ${recipientId}: ${savedNotification.id}`); + } catch (error) { + this.logger.error(`Failed to process notification for user ${recipientId}`, error); + this.metricsService.incrementCounter('notifications_failed_to_create_total', 1); + } + } + } + + private shouldSendNotification(preferences: any, event: NotificationEvent): boolean { + const category = this.mapEventTypeToCategory(event.eventType); + + if (!preferences.settings.categories[category]) { + return false; + } + + if (category === NotificationCategory.GOVERNANCE_PROPOSAL && !preferences.settings.governanceAlerts) { + return false; + } + + if (category === NotificationCategory.STAKING_CHANGE && !preferences.settings.stakingAlerts) { + return false; + } + + if (category === NotificationCategory.REWARD_DISTRIBUTION && !preferences.settings.rewardNotifications) { + return false; + } + + if (category === NotificationCategory.SECURITY_ALERT && !preferences.settings.securityAlerts) { + return false; + } + + return true; + } + + private mapEventTypeToCategory(eventType: string): NotificationCategory { + const categoryMap: Record = { + 'claim.created': NotificationCategory.NEW_CLAIM, + 'verification.assigned': NotificationCategory.VERIFICATION_ASSIGNMENT, + 'dispute.created': NotificationCategory.DISPUTE_CREATED, + 'dispute.resolved': NotificationCategory.DISPUTE_RESOLVED, + 'reputation.updated': NotificationCategory.REPUTATION_UPDATE, + 'stake.changed': NotificationCategory.STAKING_CHANGE, + 'governance.proposal.created': NotificationCategory.GOVERNANCE_PROPOSAL, + 'governance.voting.reminder': NotificationCategory.VOTING_DEADLINE, + 'reward.distributed': NotificationCategory.REWARD_DISTRIBUTION, + 'moderation.action': NotificationCategory.MODERATION_ACTION, + 'security.alert': NotificationCategory.SECURITY_ALERT, + }; + + return categoryMap[eventType] || NotificationCategory.SYSTEM_UPDATE; + } + + private createNotificationFromEvent(event: NotificationEvent, recipientId: string): Partial { + const category = this.mapEventTypeToCategory(event.eventType); + const { title, message, priority = NotificationPriority.MEDIUM } = this.extractNotificationContent(event); + + return { + userId: recipientId, + title, + message, + category, + priority, + metadata: event.payload, + sourceEvent: event, + read: false, + createdAt: new Date(), + }; + } + + private extractNotificationContent(event: NotificationEvent) { + const eventContentMap: Record = { + 'claim.created': { + title: 'New Claim Submitted', + message: `A new claim "${event.payload.title}" has been submitted to the protocol.`, + }, + 'verification.assigned': { + title: 'Verification Assignment', + message: `You have been assigned to verify claim #${event.payload.claimId}.`, + priority: NotificationPriority.HIGH, + }, + 'dispute.created': { + title: 'Dispute Filed', + message: `A dispute has been filed against claim #${event.payload.claimId}.`, + priority: NotificationPriority.HIGH, + }, + 'dispute.resolved': { + title: 'Dispute Resolved', + message: `The dispute for claim #${event.payload.claimId} has been resolved.`, + }, + 'reputation.updated': { + title: 'Reputation Update', + message: `Your reputation score has changed to ${event.payload.newScore}.`, + }, + 'stake.changed': { + title: 'Staking Update', + message: `Your staked balance has been updated to ${event.payload.newBalance}.`, + }, + 'governance.proposal.created': { + title: 'New Governance Proposal', + message: `A new proposal "${event.payload.title}" is available for voting.`, + }, + 'governance.voting.reminder': { + title: 'Voting Deadline Approaching', + message: `Voting for proposal "${event.payload.title}" ends in 24 hours.`, + priority: NotificationPriority.HIGH, + }, + 'reward.distributed': { + title: 'Reward Distributed', + message: `You have received a reward of ${event.payload.amount} ${event.payload.token}.`, + }, + 'security.alert': { + title: 'Security Alert', + message: event.payload.message, + priority: NotificationPriority.CRITICAL, + }, + }; + + return eventContentMap[event.eventType] || { + title: 'System Update', + message: 'An event has occurred in the protocol.', + }; + } + + private async queueNotificationForDelivery(notification: Notification, preferences: NotificationPreference) { + const enabledChannels = preferences.settings.enabledChannels; + + for (const channel of enabledChannels) { + await this.deliveryHistoryService.createDeliveryRecord(notification.id, channel); + + await this.notificationsQueue.add( + 'deliver-notification', + { + notificationId: notification.id, + channel, + userId: notification.userId, + }, + { + priority: this.getJobPriority(notification.priority), + attempts: 5, + backoff: { + type: 'exponential', + delay: 1000, + }, + } + ); + } + } + + private getJobPriority(priority: NotificationPriority): number { + const priorityMap = { + [NotificationPriority.CRITICAL]: 1, + [NotificationPriority.HIGH]: 2, + [NotificationPriority.MEDIUM]: 3, + [NotificationPriority.LOW]: 4, + }; + return priorityMap[priority] || 3; + } +} \ No newline at end of file diff --git a/src/notifications/services/webhook.service.ts b/src/notifications/services/webhook.service.ts new file mode 100644 index 0000000..f95ec4b --- /dev/null +++ b/src/notifications/services/webhook.service.ts @@ -0,0 +1,141 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { createHmac } from 'crypto'; +import { Notification } from '../entities/notification.entity'; +import { DeliveryResult, DeliveryStatus, WebhookConfig } from '../interfaces/notification.types'; +import { RedisService } from '../../redis/redis.service'; + +@Injectable() +export class WebhookService { + private readonly logger = new Logger(WebhookService.name); + + constructor(private readonly redisService: RedisService) {} + + async sendWebhookNotification( + notification: Notification, + webhook: WebhookConfig + ): Promise { + this.logger.debug(`Sending webhook notification to ${webhook.url} for notification ${notification.id}`); + + try { + const payload = this.createWebhookPayload(notification); + const signature = this.generateSignature(JSON.stringify(payload), webhook.secret); + + const response = await fetch(webhook.url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-TruthBounty-Signature': signature, + 'X-TruthBounty-Event': notification.category, + 'User-Agent': 'TruthBounty-Webhook/1.0', + }, + body: JSON.stringify(payload), + signal: AbortSignal.timeout(10000), + }); + + if (response.ok) { + this.logger.log(`Webhook delivered successfully to ${webhook.url}`); + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + } else { + const errorText = await response.text(); + this.logger.error(`Webhook delivery failed with status ${response.status}: ${errorText}`); + return { + success: false, + status: DeliveryStatus.FAILED, + error: `HTTP ${response.status}: ${errorText}`, + }; + } + } catch (error) { + this.logger.error(`Webhook delivery exception for ${webhook.url}`, error); + return { + success: false, + status: DeliveryStatus.FAILED, + error: error.message, + }; + } + } + + private createWebhookPayload(notification: Notification) { + return { + id: notification.id, + eventType: notification.category, + title: notification.title, + message: notification.message, + createdAt: notification.createdAt, + metadata: notification.metadata, + userId: notification.userId, + }; + } + + private generateSignature(payload: string, secret: string): string { + const hmac = createHmac('sha256', secret); + hmac.update(payload); + return `sha256=${hmac.digest('hex')}`; + } + + verifySignature(payload: string, signature: string, secret: string): boolean { + const expectedSignature = this.generateSignature(payload, secret); + return this.timingSafeCompare(expectedSignature, signature); + } + + private timingSafeCompare(a: string, b: string): boolean { + if (a.length !== b.length) return false; + + let result = 0; + for (let i = 0; i < a.length; i++) { + result |= a.charCodeAt(i) ^ b.charCodeAt(i); + } + return result === 0; + } + + async getUserWebhooks(userId: string): Promise { + const webhooks = await this.redisService.get(`webhooks:${userId}`); + return webhooks ? JSON.parse(webhooks) : []; + } + + async addWebhook(userId: string, webhook: Omit): Promise { + const webhooks = await this.getUserWebhooks(userId); + + const newWebhook: WebhookConfig = { + ...webhook, + id: crypto.randomUUID(), + createdAt: new Date(), + updatedAt: new Date(), + }; + + webhooks.push(newWebhook); + await this.redisService.set(`webhooks:${userId}`, JSON.stringify(webhooks)); + + this.logger.log(`New webhook added for user ${userId}: ${webhook.url}`); + return newWebhook; + } + + async removeWebhook(userId: string, webhookId: string): Promise { + const webhooks = await this.getUserWebhooks(userId); + const index = webhooks.findIndex(w => w.id === webhookId); + + if (index === -1) return false; + + webhooks.splice(index, 1); + await this.redisService.set(`webhooks:${userId}`, JSON.stringify(webhooks)); + + this.logger.log(`Webhook ${webhookId} removed for user ${userId}`); + return true; + } + + async updateWebhook(userId: string, webhookId: string, updates: Partial): Promise { + const webhooks = await this.getUserWebhooks(userId); + const webhook = webhooks.find(w => w.id === webhookId); + + if (!webhook) return null; + + Object.assign(webhook, updates, { updatedAt: new Date() }); + await this.redisService.set(`webhooks:${userId}`, JSON.stringify(webhooks)); + + this.logger.log(`Webhook ${webhookId} updated for user ${userId}`); + return webhook; + } +} \ No newline at end of file diff --git a/src/notifications/services/websocket.service.ts b/src/notifications/services/websocket.service.ts new file mode 100644 index 0000000..764a638 --- /dev/null +++ b/src/notifications/services/websocket.service.ts @@ -0,0 +1,157 @@ +import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; +import { Server } from 'socket.io'; +import { Notification } from '../entities/notification.entity'; +import { DeliveryResult, DeliveryStatus } from '../interfaces/notification.types'; +import { RedisService } from '../../redis/redis.service'; + +@Injectable() +export class WebSocketService implements OnModuleInit { + private readonly logger = new Logger(WebSocketService.name); + private server: Server; + private userSockets: Map> = new Map(); + + constructor(private readonly redisService: RedisService) {} + + onModuleInit() { + this.initializeWebSocketServer(); + } + + private initializeWebSocketServer() { + this.server = new Server({ + cors: { + origin: process.env.CORS_ORIGIN || '*', + credentials: true, + }, + }); + + this.setupConnectionHandlers(); + const port = process.env.WEBSOCKET_PORT || 3001; + this.server.listen(port); + + this.logger.log(`WebSocket server initialized on port ${port}`); + } + + private setupConnectionHandlers() { + this.server.on('connection', (socket) => { + this.logger.debug(`New client connected: ${socket.id}`); + + socket.on('authenticate', async (userId: string) => { + await this.registerUserSocket(userId, socket.id); + socket.join(`user:${userId}`); + this.logger.debug(`User ${userId} authenticated with socket ${socket.id}`); + }); + + socket.on('disconnect', async () => { + await this.removeUserSocket(socket.id); + this.logger.debug(`Client disconnected: ${socket.id}`); + }); + + socket.on('subscribe', (channels: string[]) => { + channels.forEach(channel => socket.join(channel)); + this.logger.debug(`Socket ${socket.id} subscribed to channels: ${channels.join(', ')}`); + }); + + socket.on('unsubscribe', (channels: string[]) => { + channels.forEach(channel => socket.leave(channel)); + this.logger.debug(`Socket ${socket.id} unsubscribed from channels: ${channels.join(', ')}`); + }); + }); + } + + private async registerUserSocket(userId: string, socketId: string) { + if (!this.userSockets.has(userId)) { + this.userSockets.set(userId, new Set()); + } + this.userSockets.get(userId).add(socketId); + + await this.redisService.sAdd(`active_sockets:${userId}`, socketId); + } + + private async removeUserSocket(socketId: string) { + for (const [userId, sockets] of this.userSockets.entries()) { + if (sockets.has(socketId)) { + sockets.delete(socketId); + await this.redisService.sRemove(`active_sockets:${userId}`, socketId); + + if (sockets.size === 0) { + this.userSockets.delete(userId); + } + break; + } + } + } + + async isUserOnline(userId: string): Promise { + const sockets = await this.redisService.sMembers(`active_sockets:${userId}`); + return sockets && sockets.length > 0; + } + + async sendToUser(userId: string, event: string, data: any): Promise { + const userRoom = `user:${userId}`; + const sockets = await this.server.in(userRoom).allSockets(); + const recipientCount = sockets.size; + + this.server.to(userRoom).emit(event, data); + this.logger.debug(`Sent ${event} to user ${userId}, ${recipientCount} recipients`); + + return recipientCount; + } + + async broadcastNotification(notification: Notification): Promise { + try { + const isOnline = await this.isUserOnline(notification.userId); + + if (!isOnline) { + return { + success: false, + status: DeliveryStatus.FAILED, + error: 'User is not connected to WebSocket', + }; + } + + const recipientCount = await this.sendToUser( + notification.userId, + 'notification:new', + notification + ); + + if (recipientCount > 0) { + return { + success: true, + status: DeliveryStatus.DELIVERED, + deliveredAt: new Date(), + }; + } else { + return { + success: false, + status: DeliveryStatus.FAILED, + error: 'No active connections found for user', + }; + } + } catch (error) { + this.logger.error(`Failed to send WebSocket notification`, error); + return { + success: false, + status: DeliveryStatus.FAILED, + error: error.message, + }; + } + } + + async broadcastGlobal(event: string, data: any): Promise { + this.server.emit(event, data); + this.logger.debug(`Broadcast global event ${event} to all connected clients`); + } + + getConnectedUsersCount(): number { + return this.userSockets.size; + } + + getTotalConnections(): number { + let total = 0; + for (const sockets of this.userSockets.values()) { + total += sockets.size; + } + return total; + } +} \ No newline at end of file