diff --git a/.freebuff/project-id b/.freebuff/project-id index e76b7ed..c7763ad 100644 --- a/.freebuff/project-id +++ b/.freebuff/project-id @@ -1 +1 @@ -f0c4cc0d-4346-4931-ae7a-b1670a0d5e69 +6873507d-0234-4bdb-b97b-da04098eb1dc diff --git a/src/app.module.ts b/src/app.module.ts index 19f6d5b..239b840 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -57,6 +57,8 @@ import { BillingModule } from "./billing/billing.module"; import { PaymentsModule } from "./payments/payments.module"; import { RateLimitingModule } from "./rate-limiting/rate-limiting.module"; import { ReconciliationModule } from "./reconciliation/reconciliation.module"; +// Modules – notifications +import { NotificationModule } from "./notifications/notification.module"; // Auth entities import { User } from "./core/user/entities/user.entity"; @@ -115,6 +117,13 @@ import { FileScanResult } from "./infrastructure/file-upload/entities/file-scan- import { ReconciliationAudit } from "./reconciliation/entities/reconciliation-audit.entity"; import { ReconciliationInvoice } from "./reconciliation/entities/reconciliation-invoice.entity"; import { StellarTransaction } from "./reconciliation/entities/stellar-transaction.entity"; +// Notification entities +import { Notification } from "./notifications/entities/notification.entity"; +import { NotificationTemplate } from "./notifications/entities/notification-template.entity"; +import { NotificationPreference } from "./notifications/entities/notification-preference.entity"; +import { NotificationAggregation } from "./notifications/entities/notification-aggregation.entity"; +import { NotificationDeliveryLog } from "./notifications/entities/notification-delivery-log.entity"; +import { NotificationAnalytics } from "./notifications/entities/notification-analytics.entity"; // Modules – webhooks import { WebhookModule } from "./infrastructure/webhooks/webhook.module"; // Modules – file upload @@ -220,6 +229,12 @@ import { TenantModuleState } from "./modules/registry/entities/tenant-module-sta ReconciliationAudit, ReconciliationInvoice, StellarTransaction, + Notification, + NotificationTemplate, + NotificationPreference, + NotificationAggregation, + NotificationDeliveryLog, + NotificationAnalytics, ], synchronize: true, logging: true, @@ -272,6 +287,7 @@ import { TenantModuleState } from "./modules/registry/entities/tenant-module-sta BillingModule, PaymentsModule, ReconciliationModule, + NotificationModule, ], controllers: [AppController], diff --git a/src/notifications/dto/notification-analytics.dto.ts b/src/notifications/dto/notification-analytics.dto.ts new file mode 100644 index 0000000..4b8fe95 --- /dev/null +++ b/src/notifications/dto/notification-analytics.dto.ts @@ -0,0 +1,62 @@ +import { + IsOptional, + IsString, + IsEnum, + IsDateString, + IsInt, + Min, + Max, +} from 'class-validator'; +import { ApiPropertyOptional } from '@nestjs/swagger'; +import { NotificationChannel } from '../entities/notification.entity'; + +export class QueryAnalyticsDto { + @ApiPropertyOptional({ description: 'Start of date range (ISO 8601)' }) + @IsOptional() + @IsDateString() + from?: string; + + @ApiPropertyOptional({ description: 'End of date range (ISO 8601)' }) + @IsOptional() + @IsDateString() + to?: string; + + @ApiPropertyOptional({ enum: ['hourly', 'daily'], default: 'daily' }) + @IsOptional() + @IsString() + granularity?: 'hourly' | 'daily' = 'daily'; + + @ApiPropertyOptional({ enum: NotificationChannel }) + @IsOptional() + @IsEnum(NotificationChannel) + channel?: NotificationChannel; + + @ApiPropertyOptional({ description: 'Filter by notification category' }) + @IsOptional() + @IsString() + category?: string; + + @ApiPropertyOptional({ default: 30, minimum: 1, maximum: 365 }) + @IsOptional() + @IsInt() + @Min(1) + @Max(365) + limit?: number = 30; +} + +export class EngagementSummaryDto { + @ApiPropertyOptional() + @IsOptional() + @IsString() + userId?: string; + + @ApiPropertyOptional({ description: 'Start of date range (ISO 8601)' }) + @IsOptional() + @IsDateString() + from?: string; + + @ApiPropertyOptional({ description: 'End of date range (ISO 8601)' }) + @IsOptional() + @IsDateString() + to?: string; +} diff --git a/src/notifications/dto/notification-preference.dto.ts b/src/notifications/dto/notification-preference.dto.ts new file mode 100644 index 0000000..d23f6a6 --- /dev/null +++ b/src/notifications/dto/notification-preference.dto.ts @@ -0,0 +1,116 @@ +import { + IsString, + IsOptional, + IsEnum, + IsBoolean, + IsInt, + Min, + Max, + IsArray, + ValidateNested, + IsObject, + IsEmail, + Matches, +} from 'class-validator'; +import { Type } from 'class-transformer'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { NotificationCategory } from '../entities/notification.entity'; +import { NotificationChannelPreference } from '../entities/notification-preference.entity'; + +export class UpdateNotificationPreferenceDto { + @ApiProperty({ enum: NotificationCategory }) + @IsEnum(NotificationCategory) + category: NotificationCategory; + + @ApiProperty({ description: 'Channel to configure (email, sms, push, webhook, in_app)' }) + @IsString() + channel: string; + + @ApiProperty({ enum: NotificationChannelPreference }) + @IsEnum(NotificationChannelPreference) + preference: NotificationChannelPreference; + + @ApiPropertyOptional({ enum: ['hourly', 'daily', 'weekly'], description: 'For digest mode' }) + @IsOptional() + @IsString() + digestFrequency?: string; + + @ApiPropertyOptional({ description: 'Quiet hours start (0-23)', minimum: 0, maximum: 23 }) + @IsOptional() + @IsInt() + @Min(0) + @Max(23) + quietHoursStart?: number; + + @ApiPropertyOptional({ description: 'Quiet hours end (0-23)', minimum: 0, maximum: 23 }) + @IsOptional() + @IsInt() + @Min(0) + @Max(23) + quietHoursEnd?: number; + + @ApiPropertyOptional({ description: 'IANA timezone (e.g., America/New_York)' }) + @IsOptional() + @IsString() + timezone?: string; + + @ApiPropertyOptional({ description: 'Email address for email channel' }) + @IsOptional() + @IsEmail() + emailAddress?: string; + + @ApiPropertyOptional({ description: 'Phone number for SMS channel (E.164 format)' }) + @IsOptional() + @IsString() + @Matches(/^\+[1-9]\d{1,14}$/) + phoneNumber?: string; + + @ApiPropertyOptional({ description: 'Device push token' }) + @IsOptional() + @IsString() + pushToken?: string; + + @ApiPropertyOptional({ description: 'Webhook callback URL' }) + @IsOptional() + @IsString() + webhookUrl?: string; + + @ApiPropertyOptional({ + enum: ['low', 'normal', 'high', 'critical'], + description: 'Minimum priority to trigger this channel', + }) + @IsOptional() + @IsString() + minPriority?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsBoolean() + active?: boolean; + + @ApiPropertyOptional() + @IsOptional() + @IsObject() + metadata?: Record; +} + +export class BulkUpdatePreferenceDto { + @ApiProperty({ type: [UpdateNotificationPreferenceDto] }) + @IsArray() + @ValidateNested({ each: true }) + @Type(() => UpdateNotificationPreferenceDto) + preferences: UpdateNotificationPreferenceDto[]; +} + +export class SetAllChannelsDto { + @ApiProperty({ enum: NotificationCategory }) + @IsEnum(NotificationCategory) + category: NotificationCategory; + + @ApiProperty({ + enum: ['all_on', 'all_off', 'in_app_only', 'essential_only'], + description: 'Preset: enable/disable all channels at once', + }) + @IsString() + preset: 'all_on' | 'all_off' | 'in_app_only' | 'essential_only'; +} diff --git a/src/notifications/dto/notification-template.dto.ts b/src/notifications/dto/notification-template.dto.ts new file mode 100644 index 0000000..ed1656e --- /dev/null +++ b/src/notifications/dto/notification-template.dto.ts @@ -0,0 +1,122 @@ +import { + IsString, + IsOptional, + IsBoolean, + IsArray, + IsObject, + MaxLength, + MinLength, +} from 'class-validator'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; + +export class CreateNotificationTemplateDto { + @ApiProperty({ example: 'portfolio_alert' }) + @IsString() + @MinLength(2) + @MaxLength(255) + name: string; + + @ApiProperty({ example: 'Portfolio value threshold alert' }) + @IsString() + @MinLength(1) + @MaxLength(255) + description: string; + + @ApiProperty({ example: 'email', description: 'Target channel' }) + @IsString() + channel: string; + + @ApiPropertyOptional({ example: 'Your portfolio has reached {{threshold}}' }) + @IsOptional() + @IsString() + subject?: string; + + @ApiProperty({ example: 'Hello {{name}}, your portfolio is now {{value}}.' }) + @IsString() + bodyTemplate: string; + + @ApiPropertyOptional({ + description: 'HTML template with Handlebars-style variables', + }) + @IsOptional() + @IsString() + htmlTemplate?: string; + + @ApiPropertyOptional({ description: 'SMS-specific template' }) + @IsOptional() + @IsString() + smsTemplate?: string; + + @ApiPropertyOptional({ + example: ['name', 'value', 'threshold'], + description: 'Template variable names', + }) + @IsOptional() + @IsArray() + variables?: string[]; + + @ApiPropertyOptional({ example: 'portfolio' }) + @IsOptional() + @IsString() + category?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsObject() + metadata?: Record; +} + +export class UpdateNotificationTemplateDto { + @ApiPropertyOptional() + @IsOptional() + @IsString() + @MaxLength(255) + description?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsString() + subject?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsString() + bodyTemplate?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsString() + htmlTemplate?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsString() + smsTemplate?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsArray() + variables?: string[]; + + @ApiPropertyOptional() + @IsOptional() + @IsBoolean() + active?: boolean; + + @ApiPropertyOptional() + @IsOptional() + @IsObject() + metadata?: Record; +} + +export class RenderTemplatePreviewDto { + @ApiProperty() + @IsString() + templateName: string; + + @ApiProperty({ + example: { name: 'Alice', value: '$15,000', threshold: '$10,000' }, + }) + @IsObject() + variables: Record; +} diff --git a/src/notifications/dto/query-notification.dto.ts b/src/notifications/dto/query-notification.dto.ts new file mode 100644 index 0000000..c6dd935 --- /dev/null +++ b/src/notifications/dto/query-notification.dto.ts @@ -0,0 +1,107 @@ +import { + IsString, + IsOptional, + IsEnum, + IsInt, + Min, + Max, + IsDateString, + IsBoolean, + IsUUID, +} from 'class-validator'; +import { Type } from 'class-transformer'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { + NotificationCategory, + NotificationChannel, + NotificationStatus, + NotificationPriority, +} from '../entities/notification.entity'; + +export class QueryNotificationHistoryDto { + @ApiProperty({ description: 'User ID' }) + @IsString() + userId: string; + + @ApiPropertyOptional({ default: 20, minimum: 1, maximum: 100 }) + @IsOptional() + @Type(() => Number) + @IsInt() + @Min(1) + @Max(100) + limit?: number = 20; + + @ApiPropertyOptional({ description: 'Offset-based pagination cursor' }) + @IsOptional() + @IsString() + cursor?: string; + + @ApiPropertyOptional({ enum: NotificationCategory, description: 'Filter by category' }) + @IsOptional() + @IsEnum(NotificationCategory) + category?: NotificationCategory; + + @ApiPropertyOptional({ enum: NotificationChannel, description: 'Filter by channel' }) + @IsOptional() + @IsEnum(NotificationChannel) + channel?: NotificationChannel; + + @ApiPropertyOptional({ + description: 'Filter by read status', + example: true, + }) + @IsOptional() + @IsBoolean() + @Type(() => Boolean) + read?: boolean; + + @ApiPropertyOptional({ enum: NotificationStatus, description: 'Filter by status' }) + @IsOptional() + @IsEnum(NotificationStatus) + status?: NotificationStatus; + + @ApiPropertyOptional({ enum: NotificationPriority, description: 'Filter by priority' }) + @IsOptional() + @IsEnum(NotificationPriority) + priority?: NotificationPriority; + + @ApiPropertyOptional({ description: 'Only notifications after this time (ISO 8601)' }) + @IsOptional() + @IsDateString() + after?: string; + + @ApiPropertyOptional({ description: 'Only notifications before this time (ISO 8601)' }) + @IsOptional() + @IsDateString() + before?: string; + + @ApiPropertyOptional({ + enum: ['createdAt', 'readAt', 'priority'], + default: 'createdAt', + }) + @IsOptional() + @IsString() + sortBy?: string = 'createdAt'; + + @ApiPropertyOptional({ enum: ['ASC', 'DESC'], default: 'DESC' }) + @IsOptional() + @IsString() + sortOrder?: 'ASC' | 'DESC' = 'DESC'; +} + +export class UnreadCountDto { + @ApiProperty() + @IsString() + userId: string; + + @ApiPropertyOptional({ enum: NotificationCategory }) + @IsOptional() + @IsEnum(NotificationCategory) + category?: NotificationCategory; +} + +export class DeleteNotificationDto { + @ApiProperty({ isArray: true, type: [String], description: 'Notification IDs to soft-delete' }) + @IsString({ each: true }) + notificationIds: string[]; +} diff --git a/src/notifications/dto/send-notification.dto.ts b/src/notifications/dto/send-notification.dto.ts new file mode 100644 index 0000000..538eaf9 --- /dev/null +++ b/src/notifications/dto/send-notification.dto.ts @@ -0,0 +1,200 @@ +import { + IsString, + IsOptional, + IsEnum, + IsObject, + IsArray, + IsBoolean, + IsDateString, + IsInt, + Min, + Max, + MaxLength, + MinLength, + ValidateNested, + IsUUID, +} from 'class-validator'; +import { Type } from 'class-transformer'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { + NotificationChannel, + NotificationCategory, + NotificationPriority, +} from '../entities/notification.entity'; + +export class SendNotificationDto { + @ApiProperty({ description: 'Recipient user ID' }) + @IsString() + userId: string; + + @ApiProperty({ example: 'Portfolio Alert' }) + @IsString() + @MinLength(1) + @MaxLength(255) + title: string; + + @ApiProperty({ example: 'Your portfolio value has dropped by 5%' }) + @IsString() + @MinLength(1) + body: string; + + @ApiPropertyOptional({ description: 'HTML version of the body' }) + @IsOptional() + @IsString() + htmlBody?: string; + + @ApiPropertyOptional({ + enum: NotificationCategory, + default: NotificationCategory.SYSTEM, + }) + @IsOptional() + @IsEnum(NotificationCategory) + category?: NotificationCategory; + + @ApiPropertyOptional({ + enum: NotificationPriority, + default: NotificationPriority.NORMAL, + }) + @IsOptional() + @IsEnum(NotificationPriority) + priority?: NotificationPriority; + + @ApiPropertyOptional({ + enum: NotificationChannel, + default: NotificationChannel.IN_APP, + description: 'Primary delivery channel', + }) + @IsOptional() + @IsEnum(NotificationChannel) + primaryChannel?: NotificationChannel; + + @ApiPropertyOptional({ + enum: NotificationChannel, + isArray: true, + description: 'All channels to deliver through', + }) + @IsOptional() + @IsArray() + @IsEnum(NotificationChannel, { each: true }) + channels?: NotificationChannel[]; + + @ApiPropertyOptional({ description: 'Template name to render' }) + @IsOptional() + @IsString() + templateName?: string; + + @ApiPropertyOptional({ + description: 'Template variables for rendering', + example: { name: 'Alice', amount: '1000' }, + }) + @IsOptional() + @IsObject() + templateVars?: Record; + + @ApiPropertyOptional({ description: 'Reference entity ID' }) + @IsOptional() + @IsString() + referenceId?: string; + + @ApiPropertyOptional({ description: 'Reference entity type' }) + @IsOptional() + @IsString() + referenceType?: string; + + /** Aggregation key to group notifications and prevent spam */ + @ApiPropertyOptional({ + description: + 'Aggregation key: notifications with the same key within the cooldown window will be collapsed', + }) + @IsOptional() + @IsString() + aggregationKey?: string; + + @ApiPropertyOptional({ + description: 'Schedule for future delivery (ISO 8601)', + }) + @IsOptional() + @IsDateString() + scheduledAt?: string; + + @ApiPropertyOptional() + @IsOptional() + @IsInt() + @Min(1) + @Max(20) + maxAttempts?: number; + + @ApiPropertyOptional() + @IsOptional() + @IsObject() + metadata?: Record; +} + +export class SendBulkNotificationDto { + @ApiProperty({ type: [SendNotificationDto] }) + @IsArray() + @ValidateNested({ each: true }) + @Type(() => SendNotificationDto) + notifications: SendNotificationDto[]; +} + +export class ScheduleNotificationDto extends SendNotificationDto { + @ApiProperty({ + description: 'ISO 8601 timestamp for scheduled delivery', + example: '2025-01-15T09:00:00.000Z', + }) + @IsDateString() + scheduledAt: string; +} + +export class CancelScheduledNotificationDto { + @ApiProperty() + @IsString() + @IsUUID() + notificationId: string; +} + +export class MarkReadDto { + @ApiProperty({ + description: 'Notification ID(s) to mark as read', + isArray: true, + type: [String], + }) + @IsArray() + @IsString({ each: true }) + notificationIds: string[]; +} + +export class MarkAllReadDto { + @ApiPropertyOptional({ enum: NotificationCategory }) + @IsOptional() + @IsEnum(NotificationCategory) + category?: NotificationCategory; + + @ApiPropertyOptional({ description: 'Only mark as read before this time' }) + @IsOptional() + @IsDateString() + before?: string; +} + +export class NotificationResponseDto { + @ApiProperty() + success: boolean; + + @ApiProperty() + notification: any; + + @ApiPropertyOptional() + message?: string; +} + +export class BulkNotificationResponseDto { + @ApiProperty() + success: boolean; + + @ApiProperty() + count: number; + + @ApiProperty({ type: [Object] }) + notifications: any[]; +} diff --git a/src/notifications/entities/notification-aggregation.entity.ts b/src/notifications/entities/notification-aggregation.entity.ts new file mode 100644 index 0000000..d10f847 --- /dev/null +++ b/src/notifications/entities/notification-aggregation.entity.ts @@ -0,0 +1,52 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + Index, +} from 'typeorm'; + +@Entity('notification_aggregations') +@Index(['userId', 'aggregationKey'], { unique: true }) +export class NotificationAggregation { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + @Index() + userId: string; + + @Column({ type: 'varchar', length: 255 }) + @Index() + aggregationKey: string; + + /** How many notifications have been collapsed into this one */ + @Column({ type: 'int', default: 1 }) + count: number; + + /** The latest notification ID that was aggregated */ + @Column({ type: 'varchar', length: 36 }) + latestNotificationId: string; + + /** When the aggregation window started */ + @Column({ type: 'timestamp' }) + windowStartedAt: Date; + + /** When the last notification was added to this aggregation */ + @Column({ type: 'timestamp' }) + lastNotificationAt: Date; + + /** Cooldown in seconds: no new notifications with this key will be sent until the cooldown expires */ + @Column({ type: 'int', default: 300 }) + cooldownSeconds: number; + + /** Whether the aggregated notification has been sent */ + @Column({ type: 'boolean', default: false }) + sent: boolean; + + @Column({ type: 'jsonb', nullable: true }) + metadata?: Record; + + @CreateDateColumn() + createdAt: Date; +} diff --git a/src/notifications/entities/notification-analytics.entity.ts b/src/notifications/entities/notification-analytics.entity.ts new file mode 100644 index 0000000..f7cc664 --- /dev/null +++ b/src/notifications/entities/notification-analytics.entity.ts @@ -0,0 +1,83 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + Index, +} from 'typeorm'; +import { NotificationChannel } from './notification.entity'; + +/** + * Aggregated analytics snapshot, computed periodically from delivery logs. + * Each row represents a time-bucket (e.g., hourly or daily) for a given channel and category. + */ +@Entity('notification_analytics') +@Index(['dateBucket', 'channel', 'category'], { unique: true }) +export class NotificationAnalytics { + @PrimaryGeneratedColumn('uuid') + id: string; + + /** Start of the time bucket (truncated to hour or day) */ + @Column({ type: 'timestamp' }) + @Index() + dateBucket: Date; + + /** Granularity: 'hourly' or 'daily' */ + @Column({ type: 'varchar', length: 20 }) + granularity: string; + + @Column({ + type: 'enum', + enum: NotificationChannel, + }) + channel: NotificationChannel; + + @Column({ type: 'varchar', length: 50, nullable: true }) + category?: string; + + /** Number of notifications attempted in this bucket */ + @Column({ type: 'int', default: 0 }) + totalSent: number; + + /** Successfully delivered */ + @Column({ type: 'int', default: 0 }) + totalDelivered: number; + + /** Failed delivery attempts */ + @Column({ type: 'int', default: 0 }) + totalFailed: number; + + /** Bounced (email hard/soft bounce, invalid phone, etc.) */ + @Column({ type: 'int', default: 0 }) + totalBounced: number; + + /** Unique users who received a notification */ + @Column({ type: 'int', default: 0 }) + uniqueRecipients: number; + + /** Notifications that were opened (email open pixel, push tap, etc.) */ + @Column({ type: 'int', default: 0 }) + totalOpened: number; + + /** Notifications where user clicked an action link */ + @Column({ type: 'int', default: 0 }) + totalClicked: number; + + /** Aggregated notifications collapsed */ + @Column({ type: 'int', default: 0 }) + totalAggregated: number; + + /** P95 delivery latency in ms */ + @Column({ type: 'int', nullable: true }) + p95DeliveryLatencyMs?: number; + + /** Average delivery latency in ms */ + @Column({ type: 'int', nullable: true }) + avgDeliveryLatencyMs?: number; + + @Column({ type: 'jsonb', nullable: true }) + metadata?: Record; + + @CreateDateColumn() + createdAt: Date; +} diff --git a/src/notifications/entities/notification-delivery-log.entity.ts b/src/notifications/entities/notification-delivery-log.entity.ts new file mode 100644 index 0000000..61c7d0b --- /dev/null +++ b/src/notifications/entities/notification-delivery-log.entity.ts @@ -0,0 +1,88 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + Index, +} from 'typeorm'; +import { NotificationChannel } from './notification.entity'; + +export enum DeliveryStatus { + PENDING = 'pending', + SENT = 'sent', + DELIVERED = 'delivered', + BOUNCED = 'bounced', + FAILED = 'failed', + CLICKED = 'clicked', + OPENED = 'opened', +} + +@Entity('notification_delivery_logs') +@Index(['notificationId', 'channel']) +@Index(['userId', 'channel']) +@Index(['status']) +@Index(['createdAt']) +export class NotificationDeliveryLog { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + notificationId: string; + + @Column() + @Index() + userId: string; + + @Column({ + type: 'enum', + enum: NotificationChannel, + }) + channel: NotificationChannel; + + @Column({ + type: 'enum', + enum: DeliveryStatus, + default: DeliveryStatus.PENDING, + }) + status: DeliveryStatus; + + @Column({ type: 'int', default: 0 }) + attemptCount: number; + + @Column({ type: 'int', default: 3 }) + maxAttempts: number; + + @Column({ type: 'text', nullable: true }) + errorMessage?: string; + + /** External provider message ID for tracking */ + @Column({ type: 'varchar', length: 255, nullable: true }) + providerMessageId?: string; + + /** Which provider was used (smtp, sendgrid, twilio, fcm, etc.) */ + @Column({ type: 'varchar', length: 100, nullable: true }) + provider?: string; + + /** Raw response from the delivery provider */ + @Column({ type: 'jsonb', nullable: true }) + providerResponse?: Record; + + @Column({ type: 'timestamp', nullable: true }) + sentAt?: Date; + + @Column({ type: 'timestamp', nullable: true }) + deliveredAt?: Date; + + @Column({ type: 'timestamp', nullable: true }) + openedAt?: Date; + + @Column({ type: 'timestamp', nullable: true }) + clickedAt?: Date; + + /** Time from send to delivery in milliseconds */ + @Column({ type: 'int', nullable: true }) + deliveryLatencyMs?: number; + + @CreateDateColumn() + createdAt: Date; +} diff --git a/src/notifications/entities/notification-preference.entity.ts b/src/notifications/entities/notification-preference.entity.ts new file mode 100644 index 0000000..740b84d --- /dev/null +++ b/src/notifications/entities/notification-preference.entity.ts @@ -0,0 +1,91 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; +import { NotificationCategory } from './notification.entity'; + +export enum NotificationChannelPreference { + ENABLED = 'enabled', + DISABLED = 'disabled', + DIGEST = 'digest', +} + +@Entity('notification_preferences') +@Index(['userId', 'category', 'channel'], { unique: true }) +export class NotificationPreference { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + @Index() + userId: string; + + @Column({ + type: 'enum', + enum: NotificationCategory, + }) + category: NotificationCategory; + + /** Which channel (email, sms, push, webhook, in_app) */ + @Column({ type: 'varchar', length: 50 }) + channel: string; + + @Column({ + type: 'enum', + enum: NotificationChannelPreference, + default: NotificationChannelPreference.ENABLED, + }) + preference: NotificationChannelPreference; + + /** For digest mode: how often to send digests (e.g., hourly, daily, weekly) */ + @Column({ type: 'varchar', length: 50, nullable: true }) + digestFrequency?: string; + + /** Quiet hours: start hour (0-23) in user's timezone */ + @Column({ type: 'int', nullable: true }) + quietHoursStart?: number; + + /** Quiet hours: end hour (0-23) in user's timezone */ + @Column({ type: 'int', nullable: true }) + quietHoursEnd?: number; + + /** User's timezone (IANA) */ + @Column({ type: 'varchar', length: 100, nullable: true }) + timezone?: string; + + /** For email: the address to send to */ + @Column({ type: 'varchar', length: 255, nullable: true }) + emailAddress?: string; + + /** For SMS: the phone number */ + @Column({ type: 'varchar', length: 50, nullable: true }) + phoneNumber?: string; + + /** For push: device token */ + @Column({ type: 'text', nullable: true }) + pushToken?: string; + + /** For webhook: callback URL */ + @Column({ type: 'varchar', length: 2048, nullable: true }) + webhookUrl?: string; + + /** Minimum priority level to trigger this channel */ + @Column({ type: 'varchar', length: 50, nullable: true }) + minPriority?: string; + + @Column({ type: 'boolean', default: true }) + active: boolean; + + @Column({ type: 'jsonb', nullable: true }) + metadata?: Record; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/notifications/entities/notification-template.entity.ts b/src/notifications/entities/notification-template.entity.ts new file mode 100644 index 0000000..72b0a20 --- /dev/null +++ b/src/notifications/entities/notification-template.entity.ts @@ -0,0 +1,54 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, +} from 'typeorm'; + +@Entity('notification_templates') +export class NotificationTemplate { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ type: 'varchar', length: 255, unique: true }) + name: string; + + @Column({ type: 'varchar', length: 255 }) + description: string; + + /** Which channel this template is for */ + @Column({ type: 'varchar', length: 50 }) + channel: string; + + @Column({ type: 'varchar', length: 500, nullable: true }) + subject?: string; + + @Column({ type: 'text', nullable: true }) + bodyTemplate: string; + + @Column({ type: 'text', nullable: true }) + htmlTemplate?: string; + + @Column({ type: 'text', nullable: true }) + smsTemplate?: string; + + /** Variables that can be used in the template: {name}, {amount}, etc. */ + @Column({ type: 'jsonb', nullable: true }) + variables?: string[]; + + @Column({ type: 'varchar', length: 50, nullable: true }) + category?: string; + + @Column({ type: 'boolean', default: true }) + active: boolean; + + @Column({ type: 'jsonb', nullable: true }) + metadata?: Record; + + @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..f83d359 --- /dev/null +++ b/src/notifications/entities/notification.entity.ts @@ -0,0 +1,173 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; + +export enum NotificationChannel { + EMAIL = 'email', + SMS = 'sms', + PUSH = 'push', + WEBHOOK = 'webhook', + IN_APP = 'in_app', +} + +export enum NotificationPriority { + LOW = 'low', + NORMAL = 'normal', + HIGH = 'high', + CRITICAL = 'critical', +} + +export enum NotificationStatus { + PENDING = 'pending', + QUEUED = 'queued', + SENDING = 'sending', + SENT = 'sent', + DELIVERED = 'delivered', + FAILED = 'failed', + CANCELLED = 'cancelled', + SCHEDULED = 'scheduled', +} + +export enum NotificationCategory { + SYSTEM = 'system', + SECURITY = 'security', + TRANSACTION = 'transaction', + PORTFOLIO = 'portfolio', + ALERT = 'alert', + MARKETING = 'marketing', + SOCIAL = 'social', + BILLING = 'billing', + COMPLIANCE = 'compliance', +} + +@Entity('notifications') +@Index(['userId', 'readAt']) +@Index(['userId', 'category']) +@Index(['userId', 'createdAt']) +@Index(['scheduledAt', 'status']) +export class Notification { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + @Index() + userId: string; + + @Column({ type: 'varchar', length: 255 }) + title: string; + + @Column({ type: 'text' }) + body: string; + + @Column({ type: 'text', nullable: true }) + htmlBody?: string; + + @Column({ + type: 'enum', + enum: NotificationCategory, + default: NotificationCategory.SYSTEM, + }) + category: NotificationCategory; + + @Column({ + type: 'enum', + enum: NotificationPriority, + default: NotificationPriority.NORMAL, + }) + priority: NotificationPriority; + + @Column({ + type: 'enum', + enum: NotificationStatus, + default: NotificationStatus.PENDING, + }) + status: NotificationStatus; + + /** Primary delivery channel */ + @Column({ + type: 'enum', + enum: NotificationChannel, + default: NotificationChannel.IN_APP, + }) + primaryChannel: NotificationChannel; + + /** All channels this notification should be delivered through */ + @Column({ type: 'simple-array', nullable: true }) + channels?: NotificationChannel[]; + + @Column({ type: 'varchar', length: 255, nullable: true }) + templateName?: string; + + @Column({ type: 'jsonb', nullable: true }) + templateVars?: Record; + + /** Optional reference ID (e.g., alert ID, transaction ID, etc.) */ + @Column({ type: 'varchar', length: 255, nullable: true }) + @Index() + referenceId?: string; + + /** Type of the referenced entity */ + @Column({ type: 'varchar', length: 100, nullable: true }) + referenceType?: string; + + @Column({ type: 'boolean', default: false }) + @Index() + read: boolean; + + @Column({ type: 'timestamp', nullable: true }) + readAt?: Date; + + /** Click-through tracking: was the notification action link clicked? */ + @Column({ type: 'boolean', default: false }) + clicked: boolean; + + @Column({ type: 'timestamp', nullable: true }) + clickedAt?: Date; + + /** If scheduled for future delivery */ + @Column({ type: 'timestamp', nullable: true }) + @Index() + scheduledAt?: Date; + + /** When delivery actually started */ + @Column({ type: 'timestamp', nullable: true }) + sentAt?: Date; + + /** When delivery was confirmed */ + @Column({ type: 'timestamp', nullable: true }) + deliveredAt?: Date; + + @Column({ type: 'int', default: 0 }) + attemptCount: number; + + @Column({ type: 'int', default: 5 }) + maxAttempts: number; + + @Column({ type: 'text', nullable: true }) + errorMessage?: string; + + /** Aggregation key: notifications with the same key within the cooldown window get collapsed */ + @Column({ type: 'varchar', length: 255, nullable: true }) + @Index() + aggregationKey?: string; + + @Column({ type: 'int', default: 0 }) + aggregationCount: number; + + @Column({ type: 'jsonb', nullable: true }) + metadata?: Record; + + @Column({ type: 'boolean', default: false }) + deleted: boolean; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/notifications/notification.controller.ts b/src/notifications/notification.controller.ts new file mode 100644 index 0000000..5349fee --- /dev/null +++ b/src/notifications/notification.controller.ts @@ -0,0 +1,337 @@ +import { + Controller, + Get, + Post, + Put, + Delete, + Param, + Body, + Query, + HttpCode, + HttpStatus, + Logger, +} from '@nestjs/common'; +import { + ApiTags, + ApiBearerAuth, + ApiOperation, + ApiResponse, + ApiParam, + ApiQuery, +} from '@nestjs/swagger'; +import { NotificationService } from './services/notification.service'; +import { NotificationTemplateService } from './services/notification-template.service'; +import { NotificationAnalyticsService } from './services/notification-analytics.service'; +import { NotificationAggregationService } from './services/notification-aggregation.service'; +import { NotificationPreferenceService } from './services/notification-preference.service'; +import { + SendNotificationDto, + SendBulkNotificationDto, + MarkReadDto, + MarkAllReadDto, +} from './dto/send-notification.dto'; +import { + UpdateNotificationPreferenceDto, + BulkUpdatePreferenceDto, + SetAllChannelsDto, +} from './dto/notification-preference.dto'; +import { + QueryNotificationHistoryDto, + DeleteNotificationDto, +} from './dto/query-notification.dto'; +import { + CreateNotificationTemplateDto, + UpdateNotificationTemplateDto, + RenderTemplatePreviewDto, +} from './dto/notification-template.dto'; +import { + QueryAnalyticsDto, + EngagementSummaryDto, +} from './dto/notification-analytics.dto'; + +@ApiTags('Notifications') +@ApiBearerAuth() +@Controller('notifications') +export class NotificationController { + private readonly logger = new Logger(NotificationController.name); + + constructor( + private readonly notificationService: NotificationService, + private readonly templateService: NotificationTemplateService, + private readonly analyticsService: NotificationAnalyticsService, + private readonly aggregationService: NotificationAggregationService, + private readonly preferenceService: NotificationPreferenceService, + ) {} + + // ─── Notification CRUD ────────────────────────────────────────────── + + @Post('send') + @HttpCode(HttpStatus.CREATED) + @ApiOperation({ summary: 'Send a single notification' }) + @ApiResponse({ status: 201, description: 'Notification sent successfully' }) + async sendNotification(@Body() dto: SendNotificationDto) { + const notification = await this.notificationService.send(dto); + return { success: true, notification }; + } + + @Post('send-bulk') + @HttpCode(HttpStatus.CREATED) + @ApiOperation({ summary: 'Send bulk notifications' }) + @ApiResponse({ status: 201, description: 'Bulk notifications sent' }) + async sendBulkNotification(@Body() dto: SendBulkNotificationDto) { + const notifications = await this.notificationService.sendBulk(dto); + return { success: true, count: notifications.length, notifications }; + } + + @Get('history/:userId') + @ApiOperation({ summary: 'Get notification history for a user' }) + @ApiParam({ name: 'userId' }) + async getHistory( + @Param('userId') userId: string, + @Query() query: QueryNotificationHistoryDto, + ) { + query.userId = userId; + const result = await this.notificationService.getHistory(query); + return { success: true, ...result }; + } + + @Get(':id') + @ApiOperation({ summary: 'Get a single notification' }) + @ApiParam({ name: 'id' }) + async getById(@Param('id') id: string) { + const notification = await this.notificationService.getById(id); + return { success: true, notification }; + } + + @Get(':id/delivery') + @ApiOperation({ summary: 'Get delivery status for a notification' }) + @ApiParam({ name: 'id' }) + async getDeliveryStatus(@Param('id') id: string) { + const logs = await this.notificationService.getDeliveryStatus(id); + return { success: true, deliveryLogs: logs }; + } + + // ─── Read / Unread Tracking ───────────────────────────────────────── + + @Post('mark-read') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Mark notifications as read' }) + async markAsRead(@Body() dto: MarkReadDto) { + const result = await this.notificationService.markAsRead(dto.notificationIds); + return { success: true, ...result }; + } + + @Post('mark-all-read') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Mark all notifications as read for a user' }) + async markAllAsRead(@Body() dto: MarkAllReadDto & { userId: string }) { + const result = await this.notificationService.markAllAsRead(dto.userId, { + category: dto.category, + before: dto.before, + }); + return { success: true, ...result }; + } + + @Post('mark-unread') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Mark notifications as unread' }) + async markAsUnread(@Body() dto: MarkReadDto) { + const result = await this.notificationService.markAsUnread(dto.notificationIds); + return { success: true, ...result }; + } + + @Post(':id/click') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Track a notification click-through' }) + @ApiParam({ name: 'id' }) + async trackClick(@Param('id') id: string) { + await this.notificationService.trackClick(id); + return { success: true }; + } + + @Get('unread-count/:userId') + @ApiOperation({ summary: 'Get unread notification count for a user' }) + @ApiParam({ name: 'userId' }) + async getUnreadCount( + @Param('userId') userId: string, + @Query('category') category?: string, + ) { + const result = await this.notificationService.getUnreadCount(userId, category); + return { success: true, ...result }; + } + + @Post('delete') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Soft-delete notifications' }) + async softDelete(@Body() dto: DeleteNotificationDto) { + const result = await this.notificationService.softDelete(dto.notificationIds); + return { success: true, ...result }; + } + + // ─── Scheduling ───────────────────────────────────────────────────── + + @Post(':id/cancel') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Cancel a scheduled notification' }) + @ApiParam({ name: 'id' }) + async cancelScheduled(@Param('id') id: string) { + const notification = await this.notificationService.cancel(id); + return { success: true, notification }; + } + + // ─── Notification Preferences ────────────────────────────────────── + + @Get('preferences/:userId') + @ApiOperation({ summary: 'Get all notification preferences for a user' }) + @ApiParam({ name: 'userId' }) + async getPreferences(@Param('userId') userId: string) { + const preferences = await this.preferenceService.getPreferences(userId); + return { success: true, preferences }; + } + + @Put('preferences/:userId') + @ApiOperation({ summary: 'Update a notification preference' }) + @ApiParam({ name: 'userId' }) + async updatePreference( + @Param('userId') userId: string, + @Body() dto: UpdateNotificationPreferenceDto, + ) { + const preference = await this.preferenceService.updatePreference(userId, dto); + return { success: true, preference }; + } + + @Put('preferences/:userId/bulk') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Bulk update notification preferences' }) + @ApiParam({ name: 'userId' }) + async bulkUpdatePreferences( + @Param('userId') userId: string, + @Body() dto: BulkUpdatePreferenceDto, + ) { + const preferences = await this.preferenceService.bulkUpdate(userId, dto.preferences); + return { success: true, count: preferences.length, preferences }; + } + + @Put('preferences/:userId/preset') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Apply a preset to all channels for a category' }) + @ApiParam({ name: 'userId' }) + async setPreset( + @Param('userId') userId: string, + @Body() dto: SetAllChannelsDto, + ) { + const preferences = await this.preferenceService.setAllChannels(userId, dto); + return { success: true, count: preferences.length }; + } + + @Post('preferences/:userId/seed') + @HttpCode(HttpStatus.CREATED) + @ApiOperation({ summary: 'Seed default notification preferences for a new user' }) + @ApiParam({ name: 'userId' }) + async seedDefaults(@Param('userId') userId: string) { + await this.preferenceService.seedDefaults(userId); + return { success: true, message: 'Default preferences seeded' }; + } + + @Delete('preferences/:userId') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Delete all notification preferences for a user' }) + @ApiParam({ name: 'userId' }) + async deleteAllPreferences(@Param('userId') userId: string) { + await this.preferenceService.deleteAll(userId); + return { success: true, message: 'All preferences deleted' }; + } + + // ─── Templates ────────────────────────────────────────────────────── + + @Get('templates/all') + @ApiOperation({ summary: 'Get all notification templates' }) + async getTemplates() { + const templates = await this.templateService.getAllTemplates(); + return { success: true, templates }; + } + + @Post('templates') + @HttpCode(HttpStatus.CREATED) + @ApiOperation({ summary: 'Create a notification template' }) + async createTemplate(@Body() dto: CreateNotificationTemplateDto) { + const template = await this.templateService.createTemplate(dto); + return { success: true, template }; + } + + @Put('templates/:name') + @ApiOperation({ summary: 'Update a notification template' }) + @ApiParam({ name: 'name' }) + async updateTemplate( + @Param('name') name: string, + @Body() dto: UpdateNotificationTemplateDto, + ) { + const template = await this.templateService.updateTemplate(name, dto); + return { success: true, template }; + } + + @Post('templates/preview') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Preview a rendered template' }) + async previewTemplate(@Body() dto: RenderTemplatePreviewDto) { + const template = await this.templateService.getTemplate(dto.templateName); + const renderedBody = this.templateService.renderBody( + dto.templateName, + template.bodyTemplate, + dto.variables, + ); + const renderedHtml = template.htmlTemplate + ? this.templateService.renderBody(dto.templateName, template.htmlTemplate, dto.variables) + : undefined; + const renderedSubject = template.subject + ? this.templateService.renderBody(dto.templateName, template.subject, dto.variables) + : undefined; + return { + success: true, + preview: { + subject: renderedSubject, + body: renderedBody, + html: renderedHtml, + }, + }; + } + + @Delete('templates/:name') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Deactivate a notification template' }) + @ApiParam({ name: 'name' }) + async deleteTemplate(@Param('name') name: string) { + await this.templateService.deleteTemplate(name); + return { success: true, message: `Template "${name}" deactivated` }; + } + + // ─── Analytics ────────────────────────────────────────────────────── + + @Get('analytics/query') + @ApiOperation({ summary: 'Query notification analytics' }) + async queryAnalytics(@Query() dto: QueryAnalyticsDto) { + const data = await this.analyticsService.queryAnalytics(dto); + return { success: true, data }; + } + + @Get('analytics/engagement') + @ApiOperation({ summary: 'Get engagement summary (delivery/open/click rates)' }) + async getEngagementSummary(@Query() dto: EngagementSummaryDto) { + const summary = await this.analyticsService.getEngagementSummary(dto); + return { success: true, ...summary }; + } + + // ─── Aggregation ──────────────────────────────────────────────────── + + @Get('aggregation/stats/:userId/:key') + @ApiOperation({ summary: 'Get aggregation stats for a user and key' }) + @ApiParam({ name: 'userId' }) + @ApiParam({ name: 'key' }) + async getAggregationStats( + @Param('userId') userId: string, + @Param('key') key: string, + ) { + const stats = await this.aggregationService.getAggregationStats(userId, key); + return { success: true, stats }; + } +} diff --git a/src/notifications/notification.module.ts b/src/notifications/notification.module.ts new file mode 100644 index 0000000..c5090f2 --- /dev/null +++ b/src/notifications/notification.module.ts @@ -0,0 +1,100 @@ +import { Module, forwardRef } from '@nestjs/common'; +import { ConfigModule, ConfigService } from '@nestjs/config'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { BullModule } from '@nestjs/bull'; +import { JwtModule } from '@nestjs/jwt'; + +// Entities +import { Notification } from './entities/notification.entity'; +import { NotificationTemplate } from './entities/notification-template.entity'; +import { NotificationPreference } from './entities/notification-preference.entity'; +import { NotificationAggregation } from './entities/notification-aggregation.entity'; +import { NotificationDeliveryLog } from './entities/notification-delivery-log.entity'; +import { NotificationAnalytics } from './entities/notification-analytics.entity'; + +// Services +import { NotificationService } from './services/notification.service'; +import { NotificationQueueService } from './services/notification-queue.service'; +import { NotificationProcessor } from './services/notification-processor.service'; +import { NotificationTemplateService } from './services/notification-template.service'; +import { NotificationPreferenceService } from './services/notification-preference.service'; +import { NotificationAggregationService } from './services/notification-aggregation.service'; +import { NotificationAnalyticsService } from './services/notification-analytics.service'; + +// Providers (Channel strategies) +import { EmailNotificationProvider } from './providers/email-notification.provider'; +import { SmsNotificationProvider } from './providers/sms-notification.provider'; +import { PushNotificationProvider } from './providers/push-notification.provider'; +import { WebhookNotificationProvider } from './providers/webhook-notification.provider'; + +// Gateway & Controller +import { NotificationGateway } from './websocket/notification.gateway'; +import { NotificationController } from './notification.controller'; + +@Module({ + imports: [ + ConfigModule, + TypeOrmModule.forFeature([ + Notification, + NotificationTemplate, + NotificationPreference, + NotificationAggregation, + NotificationDeliveryLog, + NotificationAnalytics, + ]), + BullModule.registerQueueAsync({ + name: 'notifications', + imports: [ConfigModule], + inject: [ConfigService], + useFactory: (configService: ConfigService) => ({ + redis: { + host: configService.get('REDIS_HOST', 'localhost'), + port: configService.get('REDIS_PORT', 6379), + password: configService.get('REDIS_PASSWORD'), + }, + defaultJobOptions: { + removeOnComplete: 100, + removeOnFail: 500, + }, + }), + }), + JwtModule.registerAsync({ + imports: [ConfigModule], + inject: [ConfigService], + useFactory: (configService: ConfigService) => ({ + secret: configService.get('JWT_SECRET', 'change-me-in-production'), + signOptions: { expiresIn: '24h' }, + }), + }), + ], + controllers: [NotificationController], + providers: [ + // Core services + NotificationService, + NotificationQueueService, + NotificationProcessor, + NotificationTemplateService, + NotificationPreferenceService, + NotificationAggregationService, + NotificationAnalyticsService, + + // Channel providers (Strategy pattern) + EmailNotificationProvider, + SmsNotificationProvider, + PushNotificationProvider, + WebhookNotificationProvider, + + // WebSocket gateway + NotificationGateway, + ], + exports: [ + NotificationService, + NotificationTemplateService, + NotificationPreferenceService, + NotificationAggregationService, + NotificationAnalyticsService, + NotificationQueueService, + NotificationGateway, + ], +}) +export class NotificationModule {} diff --git a/src/notifications/providers/email-notification.provider.ts b/src/notifications/providers/email-notification.provider.ts new file mode 100644 index 0000000..c20e270 --- /dev/null +++ b/src/notifications/providers/email-notification.provider.ts @@ -0,0 +1,68 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { + NotificationChannelProvider, + ChannelDeliveryPayload, + ChannelDeliveryResult, +} from './notification-channel.interface'; +import { + NotificationChannel, +} from '../entities/notification.entity'; +import { DeliveryStatus } from '../entities/notification-delivery-log.entity'; + +@Injectable() +export class EmailNotificationProvider implements NotificationChannelProvider { + readonly channel = NotificationChannel.EMAIL; + private readonly logger = new Logger(EmailNotificationProvider.name); + + constructor(private readonly configService: ConfigService) {} + + async deliver(payload: ChannelDeliveryPayload): Promise { + this.logger.log(`Delivering email notification ${payload.notificationId} to ${payload.recipientAddress}`); + + try { + // Delegate to the existing email module's SMTP/SendGrid/SES provider + // For now, simulate delivery via a configurable endpoint + const emailFrom = this.configService.get('NOTIFICATION_EMAIL_FROM', 'noreply@alian-structure.com'); + + // In production, this would use the EmailService from the email module + // or directly use nodemailer/sendgrid/SES + this.logger.log( + `Email queued: "${payload.title}" to ${payload.recipientAddress} from ${emailFrom}`, + ); + + return { + status: DeliveryStatus.SENT, + provider: 'email', + providerMessageId: `email_${payload.notificationId}_${Date.now()}`, + providerResponse: { + from: emailFrom, + to: payload.recipientAddress, + subject: payload.title, + }, + }; + } catch (error) { + this.logger.error( + `Email delivery failed for ${payload.notificationId}: ${error.message}`, + ); + return { + status: DeliveryStatus.FAILED, + errorMessage: error.message, + provider: 'email', + }; + } + } + + async validate(): Promise { + const configured = this.configService.get('EMAIL_PROVIDER') || this.configService.get('SMTP_HOST'); + return Boolean(configured); + } + + async healthCheck(): Promise { + try { + return this.validate(); + } catch { + return false; + } + } +} diff --git a/src/notifications/providers/notification-channel.interface.ts b/src/notifications/providers/notification-channel.interface.ts new file mode 100644 index 0000000..ab5e2db --- /dev/null +++ b/src/notifications/providers/notification-channel.interface.ts @@ -0,0 +1,53 @@ +import { NotificationChannel } from '../entities/notification.entity'; +import { DeliveryStatus } from '../entities/notification-delivery-log.entity'; + +/** + * Result of a channel delivery attempt. + */ +export interface ChannelDeliveryResult { + status: DeliveryStatus; + providerMessageId?: string; + provider?: string; + providerResponse?: Record; + errorMessage?: string; +} + +/** + * Payload sent to a channel provider for delivery. + */ +export interface ChannelDeliveryPayload { + notificationId: string; + userId: string; + channel: NotificationChannel; + title: string; + body: string; + htmlBody?: string; + /** Resolved contact info (email address, phone number, etc.) */ + recipientAddress: string; + /** Metadata for the channel provider */ + metadata?: Record; + priority?: string; +} + +/** + * Interface that all notification channel providers must implement. + * Follows the Strategy pattern to abstract different delivery mechanisms. + */ +export interface NotificationChannelProvider { + readonly channel: NotificationChannel; + + /** + * Attempt to deliver a notification through this channel. + */ + deliver(payload: ChannelDeliveryPayload): Promise; + + /** + * Validate that the provider is properly configured and can deliver. + */ + validate(): Promise; + + /** + * Health check for this provider. + */ + healthCheck(): Promise; +} diff --git a/src/notifications/providers/push-notification.provider.ts b/src/notifications/providers/push-notification.provider.ts new file mode 100644 index 0000000..9a948a9 --- /dev/null +++ b/src/notifications/providers/push-notification.provider.ts @@ -0,0 +1,72 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { + NotificationChannelProvider, + ChannelDeliveryPayload, + ChannelDeliveryResult, +} from './notification-channel.interface'; +import { NotificationChannel } from '../entities/notification.entity'; +import { DeliveryStatus } from '../entities/notification-delivery-log.entity'; + +@Injectable() +export class PushNotificationProvider implements NotificationChannelProvider { + readonly channel = NotificationChannel.PUSH; + private readonly logger = new Logger(PushNotificationProvider.name); + + constructor(private readonly configService: ConfigService) {} + + async deliver(payload: ChannelDeliveryPayload): Promise { + this.logger.log( + `Delivering push notification ${payload.notificationId} to ${payload.recipientAddress}`, + ); + + try { + // In production, this would use FCM (Firebase Cloud Messaging) or APNs + const pushPayload = { + token: payload.recipientAddress, + notification: { + title: payload.title, + body: payload.body, + }, + data: payload.metadata || {}, + priority: payload.priority === 'high' || payload.priority === 'critical' ? 'high' : 'normal', + }; + + this.logger.log( + `Push notification queued: "${payload.title}" to device ${payload.recipientAddress.substring(0, 12)}...`, + ); + + return { + status: DeliveryStatus.SENT, + provider: 'push', + providerMessageId: `push_${payload.notificationId}_${Date.now()}`, + providerResponse: pushPayload, + }; + } catch (error) { + this.logger.error( + `Push delivery failed for ${payload.notificationId}: ${error.message}`, + ); + return { + status: DeliveryStatus.FAILED, + errorMessage: error.message, + provider: 'push', + }; + } + } + + async validate(): Promise { + return Boolean( + this.configService.get('FCM_PROJECT_ID') || + this.configService.get('FCM_PRIVATE_KEY') || + this.configService.get('PUSH_PROVIDER'), + ); + } + + async healthCheck(): Promise { + try { + return this.validate(); + } catch { + return false; + } + } +} diff --git a/src/notifications/providers/sms-notification.provider.ts b/src/notifications/providers/sms-notification.provider.ts new file mode 100644 index 0000000..60575fa --- /dev/null +++ b/src/notifications/providers/sms-notification.provider.ts @@ -0,0 +1,70 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { + NotificationChannelProvider, + ChannelDeliveryPayload, + ChannelDeliveryResult, +} from './notification-channel.interface'; +import { NotificationChannel } from '../entities/notification.entity'; +import { DeliveryStatus } from '../entities/notification-delivery-log.entity'; + +@Injectable() +export class SmsNotificationProvider implements NotificationChannelProvider { + readonly channel = NotificationChannel.SMS; + private readonly logger = new Logger(SmsNotificationProvider.name); + + constructor(private readonly configService: ConfigService) {} + + async deliver(payload: ChannelDeliveryPayload): Promise { + this.logger.log( + `Delivering SMS notification ${payload.notificationId} to ${payload.recipientAddress}`, + ); + + try { + // In production, this would use Twilio, AWS SNS, or similar + const smsBody = payload.body.length > 160 + ? payload.body.substring(0, 157) + '...' + : payload.body; + + this.logger.log( + `SMS queued: "${smsBody}" to ${payload.recipientAddress}`, + ); + + return { + status: DeliveryStatus.SENT, + provider: 'sms', + providerMessageId: `sms_${payload.notificationId}_${Date.now()}`, + providerResponse: { + to: payload.recipientAddress, + body: smsBody, + charsUsed: smsBody.length, + }, + }; + } catch (error) { + this.logger.error( + `SMS delivery failed for ${payload.notificationId}: ${error.message}`, + ); + return { + status: DeliveryStatus.FAILED, + errorMessage: error.message, + provider: 'sms', + }; + } + } + + async validate(): Promise { + // SMS requires a configured provider (Twilio, etc.) + return Boolean( + this.configService.get('TWILIO_ACCOUNT_SID') || + this.configService.get('SMS_PROVIDER'), + ); + } + + async healthCheck(): Promise { + try { + return this.validate(); + } catch { + return false; + } + } +} diff --git a/src/notifications/providers/webhook-notification.provider.ts b/src/notifications/providers/webhook-notification.provider.ts new file mode 100644 index 0000000..f093ea2 --- /dev/null +++ b/src/notifications/providers/webhook-notification.provider.ts @@ -0,0 +1,92 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { + NotificationChannelProvider, + ChannelDeliveryPayload, + ChannelDeliveryResult, +} from './notification-channel.interface'; +import { NotificationChannel } from '../entities/notification.entity'; +import { DeliveryStatus } from '../entities/notification-delivery-log.entity'; +import * as crypto from 'crypto'; + +@Injectable() +export class WebhookNotificationProvider implements NotificationChannelProvider { + readonly channel = NotificationChannel.WEBHOOK; + private readonly logger = new Logger(WebhookNotificationProvider.name); + + constructor(private readonly configService: ConfigService) {} + + async deliver(payload: ChannelDeliveryPayload): Promise { + this.logger.log( + `Delivering webhook notification ${payload.notificationId} to ${payload.recipientAddress}`, + ); + + try { + const webhookUrl = payload.recipientAddress; + const signingKey = + payload.metadata?.signingKey || + this.configService.get('WEBHOOK_SIGNING_KEY', ''); + + const webhookPayload = JSON.stringify({ + event: 'notification', + notificationId: payload.notificationId, + userId: payload.userId, + title: payload.title, + body: payload.body, + htmlBody: payload.htmlBody, + priority: payload.priority, + timestamp: new Date().toISOString(), + metadata: payload.metadata, + }); + + // Generate HMAC signature for webhook security + const signature = signingKey + ? crypto.createHmac('sha256', signingKey).update(webhookPayload).digest('hex') + : undefined; + + const headers: Record = { + 'Content-Type': 'application/json', + 'X-Notification-ID': payload.notificationId, + 'X-Timestamp': new Date().toISOString(), + }; + if (signature) { + headers['X-Signature-256'] = `sha256=${signature}`; + } + + // In production, this would use axios to POST to the webhook URL + this.logger.log( + `Webhook queued: POST to ${webhookUrl} with ${headers['X-Signature-256'] ? 'HMAC signature' : 'no signature'}`, + ); + + return { + status: DeliveryStatus.SENT, + provider: 'webhook', + providerMessageId: `webhook_${payload.notificationId}_${Date.now()}`, + providerResponse: { + url: webhookUrl, + method: 'POST', + headers, + payloadSize: webhookPayload.length, + }, + }; + } catch (error) { + this.logger.error( + `Webhook delivery failed for ${payload.notificationId}: ${error.message}`, + ); + return { + status: DeliveryStatus.FAILED, + errorMessage: error.message, + provider: 'webhook', + }; + } + } + + async validate(): Promise { + // Webhooks are always "valid" since they use user-provided URLs + return true; + } + + async healthCheck(): Promise { + return true; + } +} diff --git a/src/notifications/services/notification-aggregation.service.ts b/src/notifications/services/notification-aggregation.service.ts new file mode 100644 index 0000000..2925069 --- /dev/null +++ b/src/notifications/services/notification-aggregation.service.ts @@ -0,0 +1,152 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository, MoreThan } from 'typeorm'; +import { NotificationAggregation } from '../entities/notification-aggregation.entity'; + +/** + * Default cooldown in seconds before a new notification with the same + * aggregation key is allowed to be sent. + */ +const DEFAULT_COOLDOWN_SECONDS = 300; // 5 minutes + +@Injectable() +export class NotificationAggregationService { + private readonly logger = new Logger(NotificationAggregationService.name); + + constructor( + @InjectRepository(NotificationAggregation) + private readonly aggregationRepo: Repository, + ) {} + + /** + * Check whether a new notification should be aggregated (suppressed) + * or allowed through. Returns { shouldSuppress, aggregationId, count }. + * + * If suppression is allowed, the caller should increment the aggregation + * count and skip sending a new notification. If the cooldown has expired, + * a new notification is allowed and the old aggregation is marked for + * delivery. + */ + async checkAggregation( + userId: string, + aggregationKey: string, + cooldownSeconds: number = DEFAULT_COOLDOWN_SECONDS, + ): Promise<{ + shouldSuppress: boolean; + aggregation?: NotificationAggregation; + currentCount: number; + }> { + if (!aggregationKey) { + return { shouldSuppress: false, currentCount: 0 }; + } + + const existing = await this.aggregationRepo.findOne({ + where: { + userId, + aggregationKey, + sent: false, + }, + }); + + if (!existing) { + // No existing aggregation window — create one + const newAgg = this.aggregationRepo.create({ + userId, + aggregationKey, + count: 1, + latestNotificationId: '', + windowStartedAt: new Date(), + lastNotificationAt: new Date(), + cooldownSeconds, + sent: false, + }); + const saved = await this.aggregationRepo.save(newAgg); + return { shouldSuppress: false, aggregation: saved, currentCount: 1 }; + } + + // Check if the cooldown has expired + const lastNotificationTime = existing.lastNotificationAt.getTime(); + const now = Date.now(); + const cooldownMs = existing.cooldownSeconds * 1000; + + if (now - lastNotificationTime < cooldownMs) { + // Cooldown is still active — suppress + existing.count += 1; + existing.lastNotificationAt = new Date(); + await this.aggregationRepo.save(existing); + this.logger.debug( + `Notification suppressed by aggregation: key=${aggregationKey}, count=${existing.count}`, + ); + return { shouldSuppress: true, aggregation: existing, currentCount: existing.count }; + } + + // Cooldown expired — mark the old aggregation as sent and start a new window + existing.sent = true; + await this.aggregationRepo.save(existing); + + const newAgg = this.aggregationRepo.create({ + userId, + aggregationKey, + count: 1, + latestNotificationId: '', + windowStartedAt: new Date(), + lastNotificationAt: new Date(), + cooldownSeconds, + sent: false, + }); + const saved = await this.aggregationRepo.save(newAgg); + return { shouldSuppress: false, aggregation: saved, currentCount: 1 }; + } + + /** + * Update the aggregation record with the actual notification ID + * that was sent. + */ + async linkNotification( + aggregationId: string, + notificationId: string, + ): Promise { + await this.aggregationRepo.update(aggregationId, { + latestNotificationId: notificationId, + }); + } + + /** + * Mark an aggregation as sent. + */ + async markSent(aggregationId: string): Promise { + await this.aggregationRepo.update(aggregationId, { sent: true }); + } + + /** + * Get aggregation stats for a user and key. + */ + async getAggregationStats( + userId: string, + aggregationKey: string, + ): Promise<{ count: number; lastNotificationAt: Date } | null> { + const agg = await this.aggregationRepo.findOne({ + where: { userId, aggregationKey, sent: false }, + order: { createdAt: 'DESC' }, + }); + if (!agg) return null; + return { count: agg.count, lastNotificationAt: agg.lastNotificationAt }; + } + + /** + * Clean up old aggregation records (older than 7 days). + */ + async cleanupOldAggregations(): Promise { + const sevenDaysAgo = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000); + const result = await this.aggregationRepo.delete({ + createdAt: MoreThan(sevenDaysAgo) ? undefined : sevenDaysAgo, + } as any); + // Alternative: raw delete + const deleteResult = await this.aggregationRepo + .createQueryBuilder() + .delete() + .where('createdAt < :date', { date: sevenDaysAgo }) + .execute(); + return deleteResult.affected || 0; + } +} diff --git a/src/notifications/services/notification-analytics.service.ts b/src/notifications/services/notification-analytics.service.ts new file mode 100644 index 0000000..eff46b4 --- /dev/null +++ b/src/notifications/services/notification-analytics.service.ts @@ -0,0 +1,209 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { Cron, CronExpression } from '@nestjs/schedule'; +import { NotificationDeliveryLog, DeliveryStatus } from '../entities/notification-delivery-log.entity'; +import { NotificationAnalytics } from '../entities/notification-analytics.entity'; +import { NotificationChannel } from '../entities/notification.entity'; +import { QueryAnalyticsDto, EngagementSummaryDto } from '../dto/notification-analytics.dto'; + +@Injectable() +export class NotificationAnalyticsService { + private readonly logger = new Logger(NotificationAnalyticsService.name); + + constructor( + @InjectRepository(NotificationDeliveryLog) + private readonly deliveryLogRepo: Repository, + @InjectRepository(NotificationAnalytics) + private readonly analyticsRepo: Repository, + ) {} + + /** + * Record a delivery event in the analytics log. + */ + async recordDeliveryEvent( + notificationId: string, + userId: string, + channel: NotificationChannel, + status: DeliveryStatus, + metadata?: Record, + ): Promise { + // The delivery log is written by the processor; this method + // updates analytics in real-time for critical events. + this.logger.debug(`Analytics event: ${channel} ${status} for notification ${notificationId}`); + } + + /** + * Query analytics data with optional filtering by channel, category, and date range. + */ + async queryAnalytics(dto: QueryAnalyticsDto): Promise { + const qb = this.analyticsRepo.createQueryBuilder('analytics'); + + if (dto.from) { + qb.andWhere('analytics.dateBucket >= :from', { from: dto.from }); + } + if (dto.to) { + qb.andWhere('analytics.dateBucket <= :to', { to: dto.to }); + } + if (dto.channel) { + qb.andWhere('analytics.channel = :channel', { channel: dto.channel }); + } + if (dto.category) { + qb.andWhere('analytics.category = :category', { category: dto.category }); + } + if (dto.granularity) { + qb.andWhere('analytics.granularity = :granularity', { granularity: dto.granularity }); + } + + qb.orderBy('analytics.dateBucket', 'ASC'); + qb.take(dto.limit || 30); + + return qb.getMany(); + } + + /** + * Get engagement summary: open rates, click rates, delivery rates. + */ + async getEngagementSummary(dto: EngagementSummaryDto): Promise<{ + totalSent: number; + totalDelivered: number; + totalOpened: number; + totalClicked: number; + deliveryRate: number; + openRate: number; + clickRate: number; + byChannel: Record; + }> { + const qb = this.analyticsRepo.createQueryBuilder('a'); + + if (dto.userId) { + // For user-specific, sum from analytics where we have recipient data + // Since analytics doesn't have userId, we fall back to delivery logs + } + if (dto.from) { + qb.andWhere('a.dateBucket >= :from', { from: dto.from }); + } + if (dto.to) { + qb.andWhere('a.dateBucket <= :to', { to: dto.to }); + } + + const data = await qb.getMany(); + + let totalSent = 0, totalDelivered = 0, totalOpened = 0, totalClicked = 0; + const byChannel: Record = {}; + + for (const row of data) { + totalSent += row.totalSent; + totalDelivered += row.totalDelivered; + totalOpened += row.totalOpened; + totalClicked += row.totalClicked; + + if (!byChannel[row.channel]) { + byChannel[row.channel] = { sent: 0, delivered: 0, opened: 0, clicked: 0 }; + } + byChannel[row.channel].sent += row.totalSent; + byChannel[row.channel].delivered += row.totalDelivered; + byChannel[row.channel].opened += row.totalOpened; + byChannel[row.channel].clicked += row.totalClicked; + } + + return { + totalSent, + totalDelivered, + totalOpened, + totalClicked, + deliveryRate: totalSent > 0 ? totalDelivered / totalSent : 0, + openRate: totalDelivered > 0 ? totalOpened / totalDelivered : 0, + clickRate: totalOpened > 0 ? totalClicked / totalOpened : 0, + byChannel, + }; + } + + /** + * Scheduled job: aggregate delivery logs into hourly analytics buckets. + * Runs every hour at minute :5. + */ + @Cron('5 * * * *') + async aggregateHourlyAnalytics(): Promise { + await this.runAggregation('hourly'); + } + + /** + * Scheduled job: aggregate hourly analytics into daily buckets. + * Runs daily at 01:05. + */ + @Cron('5 1 * * *') + async aggregateDailyAnalytics(): Promise { + await this.runAggregation('daily'); + } + + private async runAggregation(granularity: 'hourly' | 'daily'): Promise { + const truncUnit = granularity === 'hourly' ? 'hour' : 'day'; + + try { + // Get distinct (bucket, channel, category) combinations from delivery logs + const rows = await this.deliveryLogRepo + .createQueryBuilder('log') + .select([ + `date_trunc('${truncUnit}', log.createdAt) as "bucket"`, + 'log.channel as "channel"', + 'COUNT(*) as "total"', + `COUNT(*) FILTER (WHERE log.status IN ('sent', 'delivered')) as "delivered"`, + `COUNT(*) FILTER (WHERE log.status = 'failed') as "failed"`, + `COUNT(*) FILTER (WHERE log.status = 'bounced') as "bounced"`, + `COUNT(DISTINCT log.userId) as "unique"`, + `COUNT(*) FILTER (WHERE log.status = 'opened') as "opened"`, + `COUNT(*) FILTER (WHERE log.status = 'clicked') as "clicked"`, + ]) + .groupBy(`date_trunc('${truncUnit}', log.createdAt)`) + .addGroupBy('log.channel') + .getRawMany(); + + for (const row of rows) { + const bucket = new Date(row.bucket); + const channel = row.channel as NotificationChannel; + + // Upsert analytics record + const existing = await this.analyticsRepo.findOne({ + where: { dateBucket: bucket, channel, granularity }, + }); + + if (existing) { + existing.totalSent = parseInt(row.total) || 0; + existing.totalDelivered = parseInt(row.delivered) || 0; + existing.totalFailed = parseInt(row.failed) || 0; + existing.totalBounced = parseInt(row.bounced) || 0; + existing.uniqueRecipients = parseInt(row.unique) || 0; + existing.totalOpened = parseInt(row.opened) || 0; + existing.totalClicked = parseInt(row.clicked) || 0; + await this.analyticsRepo.save(existing); + } else { + const analytics = this.analyticsRepo.create({ + dateBucket: bucket, + granularity, + channel, + totalSent: parseInt(row.total) || 0, + totalDelivered: parseInt(row.delivered) || 0, + totalFailed: parseInt(row.failed) || 0, + totalBounced: parseInt(row.bounced) || 0, + uniqueRecipients: parseInt(row.unique) || 0, + totalOpened: parseInt(row.opened) || 0, + totalClicked: parseInt(row.clicked) || 0, + }); + await this.analyticsRepo.save(analytics); + } + } + + this.logger.log( + `Aggregated ${granularity} analytics for ${rows.length} bucket/channel combinations`, + ); + } catch (error) { + this.logger.error(`Analytics aggregation failed: ${error.message}`); + } + } +} diff --git a/src/notifications/services/notification-preference.service.ts b/src/notifications/services/notification-preference.service.ts new file mode 100644 index 0000000..6b3e6c6 --- /dev/null +++ b/src/notifications/services/notification-preference.service.ts @@ -0,0 +1,223 @@ +import { Injectable, Logger, NotFoundException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { + NotificationPreference, + NotificationChannelPreference, +} from '../entities/notification-preference.entity'; +import { + UpdateNotificationPreferenceDto, + SetAllChannelsDto, +} from '../dto/notification-preference.dto'; +import { NotificationCategory, NotificationChannel } from '../entities/notification.entity'; + +@Injectable() +export class NotificationPreferenceService { + private readonly logger = new Logger(NotificationPreferenceService.name); + + constructor( + @InjectRepository(NotificationPreference) + private readonly preferenceRepo: Repository, + ) {} + + /** + * Get all preferences for a user. + */ + async getPreferences(userId: string): Promise { + return this.preferenceRepo.find({ + where: { userId }, + order: { category: 'ASC', channel: 'ASC' }, + }); + } + + /** + * Get preferences for a specific user, category, and channel. + */ + async getPreference( + userId: string, + category: NotificationCategory, + channel: string, + ): Promise { + return this.preferenceRepo.findOne({ + where: { userId, category, channel }, + }); + } + + /** + * Update or create a preference for a user. + */ + async updatePreference( + userId: string, + dto: UpdateNotificationPreferenceDto, + ): Promise { + let preference = await this.preferenceRepo.findOne({ + where: { + userId, + category: dto.category, + channel: dto.channel, + }, + }); + + if (preference) { + Object.assign(preference, { + preference: dto.preference, + digestFrequency: dto.digestFrequency, + quietHoursStart: dto.quietHoursStart, + quietHoursEnd: dto.quietHoursEnd, + timezone: dto.timezone, + emailAddress: dto.emailAddress, + phoneNumber: dto.phoneNumber, + pushToken: dto.pushToken, + webhookUrl: dto.webhookUrl, + minPriority: dto.minPriority, + active: dto.active, + metadata: dto.metadata, + }); + } else { + preference = this.preferenceRepo.create({ + userId, + category: dto.category, + channel: dto.channel, + preference: dto.preference, + digestFrequency: dto.digestFrequency, + quietHoursStart: dto.quietHoursStart, + quietHoursEnd: dto.quietHoursEnd, + timezone: dto.timezone, + emailAddress: dto.emailAddress, + phoneNumber: dto.phoneNumber, + pushToken: dto.pushToken, + webhookUrl: dto.webhookUrl, + minPriority: dto.minPriority, + active: dto.active, + metadata: dto.metadata, + }); + } + + const saved = await this.preferenceRepo.save(preference); + this.logger.log( + `Preference updated: user=${userId} category=${dto.category} channel=${dto.channel} → ${dto.preference}`, + ); + return saved; + } + + /** + * Bulk update multiple preferences. + */ + async bulkUpdate( + userId: string, + dtos: UpdateNotificationPreferenceDto[], + ): Promise { + const results: NotificationPreference[] = []; + for (const dto of dtos) { + results.push(await this.updatePreference(userId, dto)); + } + this.logger.log(`Bulk preference update: ${results.length} records for user ${userId}`); + return results; + } + + /** + * Apply a preset to all channels for a category. + */ + async setAllChannels( + userId: string, + dto: SetAllChannelsDto, + ): Promise { + const channels = Object.values(NotificationChannel); + const allCategories = Object.values(NotificationCategory); + const categories = dto.category ? [dto.category] : allCategories; + + const preferences: NotificationPreference[] = []; + + for (const category of categories) { + for (const channel of channels) { + let pref: NotificationChannelPreference; + + switch (dto.preset) { + case 'all_on': + pref = NotificationChannelPreference.ENABLED; + break; + case 'all_off': + pref = NotificationChannelPreference.DISABLED; + break; + case 'in_app_only': + pref = channel === NotificationChannel.IN_APP + ? NotificationChannelPreference.ENABLED + : NotificationChannelPreference.DISABLED; + break; + case 'essential_only': + pref = channel === NotificationChannel.IN_APP || + (category === NotificationCategory.SECURITY || category === NotificationCategory.TRANSACTION) + ? NotificationChannelPreference.ENABLED + : NotificationChannelPreference.DISABLED; + break; + default: + pref = NotificationChannelPreference.ENABLED; + } + + let preference = await this.preferenceRepo.findOne({ + where: { userId, category, channel }, + }); + + if (preference) { + preference.preference = pref; + } else { + preference = this.preferenceRepo.create({ + userId, + category, + channel, + preference: pref, + }); + } + + preferences.push(await this.preferenceRepo.save(preference)); + } + } + + this.logger.log( + `Bulk channel preset "${dto.preset}" applied for user ${userId}`, + ); + return preferences; + } + + /** + * Seed default preferences for a new user. + */ + async seedDefaults(userId: string): Promise { + const categories = Object.values(NotificationCategory); + const channels = Object.values(NotificationChannel); + + for (const category of categories) { + for (const channel of channels) { + const existing = await this.preferenceRepo.findOne({ + where: { userId, category, channel }, + }); + if (!existing) { + const isEssential = + category === NotificationCategory.SECURITY || + category === NotificationCategory.TRANSACTION || + channel === NotificationChannel.IN_APP; + + await this.preferenceRepo.save( + this.preferenceRepo.create({ + userId, + category, + channel, + preference: isEssential + ? NotificationChannelPreference.ENABLED + : NotificationChannelPreference.DISABLED, + }), + ); + } + } + } + this.logger.log(`Default preferences seeded for user ${userId}`); + } + + /** + * Delete all preferences for a user. + */ + async deleteAll(userId: string): Promise { + await this.preferenceRepo.delete({ userId }); + this.logger.log(`All preferences deleted for user ${userId}`); + } +} diff --git a/src/notifications/services/notification-processor.service.ts b/src/notifications/services/notification-processor.service.ts new file mode 100644 index 0000000..659106f --- /dev/null +++ b/src/notifications/services/notification-processor.service.ts @@ -0,0 +1,236 @@ +import { Process, Processor, OnQueueFailed, OnQueueCompleted } from '@nestjs/bull'; +import { Logger } from '@nestjs/common'; +import { Job } from 'bull'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { Notification, NotificationChannel, NotificationStatus } from '../entities/notification.entity'; +import { NotificationDeliveryLog, DeliveryStatus } from '../entities/notification-delivery-log.entity'; +import { NotificationPreference, NotificationChannelPreference } from '../entities/notification-preference.entity'; +import { NotificationChannelProvider } from '../providers/notification-channel.interface'; + +interface NotificationJobData { + notificationId: string; +} + +@Processor('notifications') +export class NotificationProcessor { + private readonly logger = new Logger(NotificationProcessor.name); + + private providers = new Map(); + + constructor( + @InjectRepository(Notification) + private readonly notificationRepo: Repository, + @InjectRepository(NotificationDeliveryLog) + private readonly deliveryLogRepo: Repository, + @InjectRepository(NotificationPreference) + private readonly preferenceRepo: Repository, + // Providers injected via registerProvider + ) {} + + /** + * Register a channel provider (called by the module on init). + */ + registerProvider(provider: NotificationChannelProvider): void { + this.providers.set(provider.channel, provider); + } + + @Process('process-notification') + async processNotification(job: Job): Promise { + const { notificationId } = job.data; + this.logger.log(`Processing notification: ${notificationId}`); + + const notification = await this.notificationRepo.findOne({ + where: { id: notificationId }, + }); + if (!notification) { + this.logger.warn(`Notification not found: ${notificationId}`); + return; + } + + if (notification.status === NotificationStatus.CANCELLED) { + this.logger.log(`Notification cancelled, skipping: ${notificationId}`); + return; + } + + // Mark as sending + await this.notificationRepo.update(notificationId, { + status: NotificationStatus.SENDING, + attemptCount: notification.attemptCount + 1, + }); + + const channels = notification.channels || [notification.primaryChannel]; + + for (const channel of channels) { + // Check user preferences + const preference = await this.preferenceRepo.findOne({ + where: { + userId: notification.userId, + channel, + category: notification.category, + }, + }); + + // Skip if the user has disabled this channel + if (preference && preference.preference === NotificationChannelPreference.DISABLED) { + this.logger.debug( + `Skipping ${channel} for user ${notification.userId}: channel disabled`, + ); + continue; + } + + // Check quiet hours + if (preference?.quietHoursStart !== null && preference?.quietHoursEnd !== null) { + const userHour = this.getUserLocalHour(preference.timezone); + if (userHour !== null && + preference.quietHoursStart !== null && + preference.quietHoursEnd !== null && + userHour >= preference.quietHoursStart && + userHour < preference.quietHoursEnd) { + this.logger.debug( + `Skipping ${channel} for user ${notification.userId}: quiet hours`, + ); + continue; + } + } + + // Check minimum priority + if (preference?.minPriority && preference.minPriority !== 'low') { + const priorityOrder = ['low', 'normal', 'high', 'critical']; + const minIdx = priorityOrder.indexOf(preference.minPriority); + const notifIdx = priorityOrder.indexOf(notification.priority); + if (notifIdx < minIdx) { + this.logger.debug( + `Skipping ${channel} for user ${notification.userId}: priority below threshold`, + ); + continue; + } + } + + // Resolve provider + const provider = this.providers.get(channel); + if (!provider) { + this.logger.warn(`No provider registered for channel ${channel}`); + continue; + } + + // Resolve recipient address from preferences + const recipientAddress = this.resolveRecipientAddress(preference, channel); + if (!recipientAddress) { + this.logger.warn( + `No recipient address for user ${notification.userId} on channel ${channel}`, + ); + continue; + } + + // Create delivery log entry + const deliveryLog = this.deliveryLogRepo.create({ + notificationId, + userId: notification.userId, + channel, + status: DeliveryStatus.PENDING, + maxAttempts: notification.maxAttempts, + }); + const savedLog = await this.deliveryLogRepo.save(deliveryLog); + + // Deliver + try { + const result = await provider.deliver({ + notificationId, + userId: notification.userId, + channel, + title: notification.title, + body: notification.body, + htmlBody: notification.htmlBody, + recipientAddress, + metadata: notification.metadata, + priority: notification.priority, + }); + + await this.deliveryLogRepo.update(savedLog.id, { + status: result.status, + providerMessageId: result.providerMessageId, + provider: result.provider, + providerResponse: result.providerResponse, + errorMessage: result.errorMessage, + sentAt: result.status === DeliveryStatus.SENT ? new Date() : undefined, + deliveredAt: result.status === DeliveryStatus.DELIVERED ? new Date() : undefined, + }); + } catch (error) { + await this.deliveryLogRepo.update(savedLog.id, { + status: DeliveryStatus.FAILED, + errorMessage: error.message, + }); + } + } + + // Update notification status + const allLogs = await this.deliveryLogRepo.find({ + where: { notificationId }, + }); + const allDelivered = allLogs.length > 0 && allLogs.every(l => + l.status === DeliveryStatus.DELIVERED || l.status === DeliveryStatus.SENT, + ); + const anyFailed = allLogs.some(l => l.status === DeliveryStatus.FAILED); + + const newStatus = allDelivered + ? NotificationStatus.DELIVERED + : anyFailed && notification.attemptCount >= notification.maxAttempts + ? NotificationStatus.FAILED + : NotificationStatus.QUEUED; + + await this.notificationRepo.update(notificationId, { + status: newStatus, + sentAt: new Date(), + deliveredAt: allDelivered ? new Date() : undefined, + }); + } + + @OnQueueFailed() + onJobFailed(job: Job, error: Error): void { + this.logger.error( + `Notification job ${job.id} failed for ${job.data.notificationId}: ${error.message}`, + ); + } + + @OnQueueCompleted() + onJobCompleted(job: Job): void { + this.logger.log(`Notification job ${job.id} completed for ${job.data.notificationId}`); + } + + private resolveRecipientAddress( + preference: NotificationPreference | null, + channel: NotificationChannel, + ): string | null { + if (!preference) return null; + switch (channel) { + case NotificationChannel.EMAIL: + return preference.emailAddress || null; + case NotificationChannel.SMS: + return preference.phoneNumber || null; + case NotificationChannel.PUSH: + return preference.pushToken || null; + case NotificationChannel.WEBHOOK: + return preference.webhookUrl || null; + case NotificationChannel.IN_APP: + return preference.userId || null; + default: + return null; + } + } + + private getUserLocalHour(timezone?: string | null): number | null { + if (!timezone) return null; + try { + const now = new Date(); + const formatter = new Intl.DateTimeFormat('en-US', { + hour: 'numeric', + hour12: false, + timeZone: timezone, + }); + return parseInt(formatter.format(now), 10); + } catch { + return null; + } + } +} diff --git a/src/notifications/services/notification-queue.service.ts b/src/notifications/services/notification-queue.service.ts new file mode 100644 index 0000000..4becfb0 --- /dev/null +++ b/src/notifications/services/notification-queue.service.ts @@ -0,0 +1,113 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectQueue } from '@nestjs/bull'; +import { Queue } from 'bull'; + +@Injectable() +export class NotificationQueueService { + private readonly logger = new Logger(NotificationQueueService.name); + + constructor( + @InjectQueue('notifications') private readonly notificationQueue: Queue, + ) {} + + /** + * Enqueue a notification for async processing. + */ + async enqueueNotification( + notificationId: string, + options?: { delay?: number; priority?: number }, + ): Promise { + const jobOptions: any = { + removeOnComplete: 100, + removeOnFail: 500, + }; + + if (options?.delay) { + jobOptions.delay = options.delay; + } + if (options?.priority !== undefined) { + jobOptions.priority = options.priority; + } + + await this.notificationQueue.add('process-notification', { notificationId }, jobOptions); + this.logger.log(`Notification enqueued: ${notificationId}`); + } + + /** + * Enqueue a scheduled notification for future delivery. + */ + async enqueueScheduledNotification( + notificationId: string, + scheduledAt: Date, + ): Promise { + const delay = scheduledAt.getTime() - Date.now(); + if (delay <= 0) { + // Already past due — process immediately + await this.enqueueNotification(notificationId); + return; + } + + await this.notificationQueue.add( + 'process-notification', + { notificationId }, + { + delay, + removeOnComplete: 100, + removeOnFail: 500, + }, + ); + this.logger.log( + `Notification scheduled: ${notificationId} for ${scheduledAt.toISOString()} (delay: ${delay}ms)`, + ); + } + + /** + * Cancel a pending scheduled notification. + */ + async cancelNotification(notificationId: string): Promise { + const jobs = await this.notificationQueue.getJobs(['delayed', 'waiting', 'active']); + for (const job of jobs) { + if (job.data?.notificationId === notificationId) { + await job.remove(); + this.logger.log(`Notification cancelled: ${notificationId}`); + return true; + } + } + return false; + } + + /** + * Get queue metrics. + */ + async getQueueMetrics(): Promise<{ + waiting: number; + active: number; + completed: number; + failed: number; + delayed: number; + }> { + const [waiting, active, completed, failed, delayed] = await Promise.all([ + this.notificationQueue.getWaitingCount(), + this.notificationQueue.getActiveCount(), + this.notificationQueue.getCompletedCount(), + this.notificationQueue.getFailedCount(), + this.notificationQueue.getDelayedCount(), + ]); + + return { waiting, active, completed, failed, delayed }; + } + + /** + * Retry failed jobs. + */ + async retryFailed(): Promise { + const failedJobs = await this.notificationQueue.getFailed(); + let retried = 0; + for (const job of failedJobs) { + await job.retry(); + retried++; + } + this.logger.log(`Retried ${retried} failed notification jobs`); + return retried; + } +} diff --git a/src/notifications/services/notification-template.service.ts b/src/notifications/services/notification-template.service.ts new file mode 100644 index 0000000..6b679a1 --- /dev/null +++ b/src/notifications/services/notification-template.service.ts @@ -0,0 +1,136 @@ +import { Injectable, Logger, NotFoundException, ConflictException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { NotificationTemplate } from '../entities/notification-template.entity'; +import { + CreateNotificationTemplateDto, + UpdateNotificationTemplateDto, +} from '../dto/notification-template.dto'; + +@Injectable() +export class NotificationTemplateService { + private readonly logger = new Logger(NotificationTemplateService.name); + + constructor( + @InjectRepository(NotificationTemplate) + private readonly templateRepo: Repository, + ) {} + + async createTemplate( + dto: CreateNotificationTemplateDto, + ): Promise { + const existing = await this.templateRepo.findOne({ + where: { name: dto.name }, + }); + if (existing) { + throw new ConflictException(`Template "${dto.name}" already exists`); + } + + const template = this.templateRepo.create({ + name: dto.name, + description: dto.description, + channel: dto.channel, + subject: dto.subject, + bodyTemplate: dto.bodyTemplate, + htmlTemplate: dto.htmlTemplate, + smsTemplate: dto.smsTemplate, + variables: dto.variables, + category: dto.category, + metadata: dto.metadata, + }); + + const saved = await this.templateRepo.save(template); + this.logger.log(`Template created: ${saved.name}`); + return saved; + } + + async updateTemplate( + name: string, + dto: UpdateNotificationTemplateDto, + ): Promise { + const template = await this.templateRepo.findOne({ where: { name } }); + if (!template) { + throw new NotFoundException(`Template "${name}" not found`); + } + + Object.assign(template, dto); + const saved = await this.templateRepo.save(template); + this.logger.log(`Template updated: ${saved.name}`); + return saved; + } + + async getTemplate(name: string): Promise { + const template = await this.templateRepo.findOne({ where: { name } }); + if (!template) { + throw new NotFoundException(`Template "${name}" not found`); + } + return template; + } + + async getAllTemplates( + activeOnly = true, + ): Promise { + const where = activeOnly ? { active: true } : {}; + return this.templateRepo.find({ where, order: { name: 'ASC' } }); + } + + async deleteTemplate(name: string): Promise { + const template = await this.templateRepo.findOne({ where: { name } }); + if (!template) { + throw new NotFoundException(`Template "${name}" not found`); + } + template.active = false; + await this.templateRepo.save(template); + this.logger.log(`Template deactivated: ${name}`); + } + + /** + * Render a template body by replacing {{variable}} placeholders. + * Supports {{variable}}, {{{variable}}} (triple-brace for raw HTML), and + * {{#if variable}}...{{/if}} conditional blocks. + */ + renderBody( + templateName: string, + bodyTemplate: string, + variables: Record = {}, + ): string { + let rendered = bodyTemplate; + + // Handle {{#if variable}}...{{/if}} blocks + rendered = rendered.replace( + /\{\{#if\s+(\w+)\}\}([\s\S]*?)\{\{\/if\}\}/g, + (_match, varName, content) => { + return variables[varName] ? content : ''; + }, + ); + + // Handle {{variable}} placeholders + rendered = rendered.replace(/\{\{(\w+)\}\}/g, (_match, varName) => { + return variables[varName] !== undefined + ? String(variables[varName]) + : `{{${varName}}}`; + }); + + return rendered; + } + + /** + * Render the subject line of a notification. + */ + renderSubject( + templateName: string, + variables: Record = {}, + ): string { + const template = this.templateNameCache.get(templateName); + if (!template?.subject) return templateName; + return this.renderBody(templateName, template.subject, variables); + } + + // Simple in-memory cache for synchronous subject rendering + private templateNameCache = new Map(); + + async cacheTemplate(name: string): Promise { + const template = await this.templateRepo.findOne({ where: { name } }); + if (template) this.templateNameCache.set(name, template); + } +} diff --git a/src/notifications/services/notification.service.ts b/src/notifications/services/notification.service.ts new file mode 100644 index 0000000..392e480 --- /dev/null +++ b/src/notifications/services/notification.service.ts @@ -0,0 +1,453 @@ +import { Injectable, Logger, NotFoundException, BadRequestException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { EventEmitter2 } from '@nestjs/event-emitter'; +import { + Notification, + NotificationStatus, + NotificationChannel, + NotificationPriority, +} from '../entities/notification.entity'; +import { NotificationDeliveryLog, DeliveryStatus } from '../entities/notification-delivery-log.entity'; +import { SendNotificationDto, SendBulkNotificationDto } from '../dto/send-notification.dto'; +import { QueryNotificationHistoryDto } from '../dto/query-notification.dto'; +import { NotificationTemplateService } from './notification-template.service'; +import { NotificationAggregationService } from './notification-aggregation.service'; +import { NotificationQueueService } from './notification-queue.service'; + +@Injectable() +export class NotificationService { + private readonly logger = new Logger(NotificationService.name); + + constructor( + @InjectRepository(Notification) + private readonly notificationRepo: Repository, + @InjectRepository(NotificationDeliveryLog) + private readonly deliveryLogRepo: Repository, + private readonly templateService: NotificationTemplateService, + private readonly aggregationService: NotificationAggregationService, + private readonly queueService: NotificationQueueService, + private readonly eventEmitter: EventEmitter2, + ) {} + + /** + * Send a single notification, optionally through multiple channels. + */ + async send(dto: SendNotificationDto): Promise { + // Check aggregation + if (dto.aggregationKey) { + const aggResult = await this.aggregationService.checkAggregation( + dto.userId, + dto.aggregationKey, + ); + if (aggResult.shouldSuppress) { + this.logger.log( + `Notification suppressed by aggregation: key=${dto.aggregationKey}, count=${aggResult.currentCount}`, + ); + // Still create a notification record, but mark it as suppressed via aggregation + const suppressed = this.notificationRepo.create({ + userId: dto.userId, + title: dto.title, + body: dto.body, + htmlBody: dto.htmlBody, + category: dto.category, + priority: dto.priority, + primaryChannel: dto.primaryChannel || NotificationChannel.IN_APP, + channels: dto.channels, + templateName: dto.templateName, + templateVars: dto.templateVars, + referenceId: dto.referenceId, + referenceType: dto.referenceType, + aggregationKey: dto.aggregationKey, + aggregationCount: aggResult.currentCount, + status: NotificationStatus.CANCELLED, + metadata: { suppressedByAggregation: true }, + }); + return this.notificationRepo.save(suppressed); + } + // Link aggregation to the new notification after creation + } + + // Render template if provided + let title = dto.title; + let body = dto.body; + let htmlBody = dto.htmlBody; + + if (dto.templateName && dto.templateVars) { + try { + const template = await this.templateService.getTemplate(dto.templateName); + body = this.templateService.renderBody( + dto.templateName, + template.bodyTemplate, + dto.templateVars, + ); + if (template.htmlTemplate) { + htmlBody = this.templateService.renderBody( + dto.templateName, + template.htmlTemplate, + dto.templateVars, + ); + } + if (template.subject) { + title = this.templateService.renderBody( + dto.templateName, + template.subject, + dto.templateVars, + ); + } + } catch (error) { + this.logger.warn( + `Template "${dto.templateName}" not found, using raw title/body`, + ); + } + } + + // Determine channels + const channels = dto.channels || [dto.primaryChannel || NotificationChannel.IN_APP]; + + // Create the notification + const notification = this.notificationRepo.create({ + userId: dto.userId, + title, + body, + htmlBody, + category: dto.category, + priority: dto.priority || NotificationPriority.NORMAL, + primaryChannel: dto.primaryChannel || NotificationChannel.IN_APP, + channels, + templateName: dto.templateName, + templateVars: dto.templateVars, + referenceId: dto.referenceId, + referenceType: dto.referenceType, + aggregationKey: dto.aggregationKey, + aggregationCount: dto.aggregationKey ? 1 : 0, + maxAttempts: dto.maxAttempts || 5, + scheduledAt: dto.scheduledAt ? new Date(dto.scheduledAt) : undefined, + status: dto.scheduledAt ? NotificationStatus.SCHEDULED : NotificationStatus.QUEUED, + metadata: dto.metadata, + }); + + const saved = await this.notificationRepo.save(notification); + + // Link aggregation + if (dto.aggregationKey) { + // Re-fetch to get the latest aggregation record + const aggStats = await this.aggregationService.getAggregationStats( + dto.userId, + dto.aggregationKey, + ); + if (aggStats) { + // The aggregation was already created by checkAggregation, update the notification ID + const aggRecords = await this.aggregationService['aggregationRepo'].find({ + where: { userId: dto.userId, aggregationKey: dto.aggregationKey, sent: false }, + order: { createdAt: 'DESC' }, + take: 1, + }); + if (aggRecords.length > 0) { + await this.aggregationService.linkNotification(aggRecords[0].id, saved.id); + } + } + } + + // Queue for delivery + if (!dto.scheduledAt) { + await this.queueService.enqueueNotification(saved.id, { + priority: this.priorityToNumber(saved.priority), + }); + } else { + await this.queueService.enqueueScheduledNotification( + saved.id, + new Date(dto.scheduledAt), + ); + } + + this.logger.log(`Notification created: ${saved.id} for user ${dto.userId}`); + return saved; + } + + /** + * Send notifications to multiple recipients. + */ + async sendBulk(dto: SendBulkNotificationDto): Promise { + const results: Notification[] = []; + for (const notifDto of dto.notifications) { + try { + results.push(await this.send(notifDto)); + } catch (error) { + this.logger.error( + `Bulk notification failed for user ${notifDto.userId}: ${error.message}`, + ); + } + } + this.logger.log(`Bulk send: ${results.length}/${dto.notifications.length} succeeded`); + return results; + } + + /** + * Cancel a scheduled notification. + */ + async cancel(notificationId: string): Promise { + const notification = await this.notificationRepo.findOne({ + where: { id: notificationId }, + }); + if (!notification) { + throw new NotFoundException(`Notification ${notificationId} not found`); + } + if (notification.status !== NotificationStatus.SCHEDULED) { + throw new BadRequestException( + `Cannot cancel notification in status ${notification.status}`, + ); + } + + await this.queueService.cancelNotification(notificationId); + notification.status = NotificationStatus.CANCELLED; + return this.notificationRepo.save(notification); + } + + /** + * Get notification history for a user with pagination. + */ + async getHistory(dto: QueryNotificationHistoryDto): Promise<{ + data: Notification[]; + total: number; + nextCursor: string | null; + }> { + const qb = this.notificationRepo + .createQueryBuilder('n') + .where('n.userId = :userId', { userId: dto.userId }) + .andWhere('n.deleted = false'); + + if (dto.category) { + qb.andWhere('n.category = :category', { category: dto.category }); + } + if (dto.channel) { + qb.andWhere( + '(n.primaryChannel = :channel OR :channel = ANY(n.channels))', + { channel: dto.channel }, + ); + } + if (dto.read !== undefined) { + if (dto.read) { + qb.andWhere('n.read = true'); + } else { + qb.andWhere('n.read = false'); + } + } + if (dto.status) { + qb.andWhere('n.status = :status', { status: dto.status }); + } + if (dto.priority) { + qb.andWhere('n.priority = :priority', { priority: dto.priority }); + } + if (dto.after) { + qb.andWhere('n.createdAt > :after', { after: dto.after }); + } + if (dto.before) { + qb.andWhere('n.createdAt < :before', { before: dto.before }); + } + + // Cursor-based pagination + if (dto.cursor) { + const decoded = Buffer.from(dto.cursor, 'base64').toString('utf8'); + try { + const { createdAt, id } = JSON.parse(decoded); + qb.andWhere( + '(n.createdAt < :cursorCreatedAt OR (n.createdAt = :cursorCreatedAt AND n.id < :cursorId))', + { cursorCreatedAt: createdAt, cursorId: id }, + ); + } catch { + throw new BadRequestException('Invalid cursor'); + } + } + + const sortOrder = dto.sortOrder || 'DESC'; + const sortBy = dto.sortBy || 'createdAt'; + qb.orderBy(`n.${sortBy}`, sortOrder).addOrderBy('n.id', sortOrder); + + // Fetch one extra to determine if there are more results + const limit = dto.limit || 20; + qb.take(limit + 1); + + const results = await qb.getMany(); + const hasMore = results.length > limit; + if (hasMore) results.pop(); + + const nextCursor = + hasMore && results.length > 0 + ? Buffer.from( + JSON.stringify({ + createdAt: results[results.length - 1].createdAt, + id: results[results.length - 1].id, + }), + ).toString('base64') + : null; + + const total = await qb.getCount(); + + return { data: results, total, nextCursor }; + } + + /** + * Get a single notification by ID. + */ + async getById(notificationId: string): Promise { + const notification = await this.notificationRepo.findOne({ + where: { id: notificationId }, + }); + if (!notification) { + throw new NotFoundException(`Notification ${notificationId} not found`); + } + return notification; + } + + /** + * Mark one or more notifications as read. + */ + async markAsRead(notificationIds: string[]): Promise<{ updated: number }> { + const result = await this.notificationRepo + .createQueryBuilder() + .update() + .set({ read: true, readAt: new Date() }) + .where('id IN (:...ids)', { ids: notificationIds }) + .andWhere('read = false') + .execute(); + + this.logger.log(`Marked ${result.affected} notifications as read`); + return { updated: result.affected || 0 }; + } + + /** + * Mark all notifications as read for a user (optionally filtered by category or time). + */ + async markAllAsRead( + userId: string, + options?: { category?: string; before?: string }, + ): Promise<{ updated: number }> { + const qb = this.notificationRepo + .createQueryBuilder() + .update() + .set({ read: true, readAt: new Date() }) + .where('userId = :userId', { userId }) + .andWhere('read = false'); + + if (options?.category) { + qb.andWhere('category = :category', { category: options.category }); + } + if (options?.before) { + qb.andWhere('createdAt <= :before', { before: options.before }); + } + + const result = await qb.execute(); + this.logger.log(`Marked ${result.affected} notifications as read for user ${userId}`); + return { updated: result.affected || 0 }; + } + + /** + * Mark one or more notifications as unread. + */ + async markAsUnread(notificationIds: string[]): Promise<{ updated: number }> { + const result = await this.notificationRepo + .createQueryBuilder() + .update() + .set({ read: false, readAt: null }) + .where('id IN (:...ids)', { ids: notificationIds }) + .andWhere('read = true') + .execute(); + + return { updated: result.affected || 0 }; + } + + /** + * Track notification click-through. + */ + async trackClick(notificationId: string): Promise { + await this.notificationRepo.update(notificationId, { + clicked: true, + clickedAt: new Date(), + }); + // Update delivery log + await this.deliveryLogRepo.update( + { notificationId }, + { status: DeliveryStatus.CLICKED, clickedAt: new Date() }, + ); + } + + /** + * Soft-delete notifications. + */ + async softDelete(notificationIds: string[]): Promise<{ deleted: number }> { + const result = await this.notificationRepo + .createQueryBuilder() + .update() + .set({ deleted: true }) + .where('id IN (:...ids)', { ids: notificationIds }) + .execute(); + + return { deleted: result.affected || 0 }; + } + + /** + * Get unread count for a user, optionally filtered by category. + */ + async getUnreadCount( + userId: string, + category?: string, + ): Promise<{ total: number; byCategory: Record }> { + const qb = this.notificationRepo + .createQueryBuilder('n') + .select('n.category', 'category') + .addSelect('COUNT(*)', 'count') + .where('n.userId = :userId', { userId }) + .andWhere('n.read = false') + .andWhere('n.deleted = false'); + + if (category) { + qb.andWhere('n.category = :category', { category }); + } + + qb.groupBy('n.category'); + const rows = await qb.getRawMany(); + + const byCategory: Record = {}; + let total = 0; + for (const row of rows) { + const count = parseInt(row.count, 10); + byCategory[row.category] = count; + total += count; + } + + return { total, byCategory }; + } + + /** + * Get delivery status for a specific notification. + */ + async getDeliveryStatus( + notificationId: string, + ): Promise { + return this.deliveryLogRepo.find({ + where: { notificationId }, + order: { createdAt: 'ASC' }, + }); + } + + /** + * Emit a real-time event when a new in-app notification is created. + */ + private emitRealtimeEvent(notification: Notification): void { + this.eventEmitter.emit('notification.new', { + userId: notification.userId, + notification, + }); + } + + private priorityToNumber(priority: string): number { + const map: Record = { + critical: 1, + high: 2, + normal: 3, + low: 4, + }; + return map[priority] || 3; + } +} diff --git a/src/notifications/test/notification-aggregation.service.spec.ts b/src/notifications/test/notification-aggregation.service.spec.ts new file mode 100644 index 0000000..9842a26 --- /dev/null +++ b/src/notifications/test/notification-aggregation.service.spec.ts @@ -0,0 +1,174 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { NotificationAggregationService } from '../services/notification-aggregation.service'; +import { NotificationAggregation } from '../entities/notification-aggregation.entity'; + +describe('NotificationAggregationService', () => { + let service: NotificationAggregationService; + let repo: Repository; + + const mockRepo = { + create: jest.fn(), + save: jest.fn(), + find: jest.fn(), + findOne: jest.fn(), + update: jest.fn(), + delete: jest.fn(), + createQueryBuilder: jest.fn(() => ({ + delete: jest.fn().mockReturnThis(), + where: jest.fn().mockReturnThis(), + execute: jest.fn().mockResolvedValue({ affected: 0 }), + })), + }; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + NotificationAggregationService, + { provide: getRepositoryToken(NotificationAggregation), useValue: mockRepo }, + ], + }).compile(); + + service = module.get(NotificationAggregationService); + repo = module.get(getRepositoryToken(NotificationAggregation)); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + describe('checkAggregation', () => { + it('should allow notification through when no existing aggregation', async () => { + mockRepo.findOne.mockResolvedValue(null); + const newAgg = { + id: 'agg-1', + userId: 'user-1', + aggregationKey: 'key-1', + count: 1, + sent: false, + }; + mockRepo.create.mockReturnValue(newAgg); + mockRepo.save.mockResolvedValue(newAgg); + + const result = await service.checkAggregation('user-1', 'key-1'); + + expect(result.shouldSuppress).toBe(false); + expect(result.currentCount).toBe(1); + expect(mockRepo.create).toHaveBeenCalled(); + expect(mockRepo.save).toHaveBeenCalled(); + }); + + it('should suppress when cooldown is still active', async () => { + const existing = { + id: 'agg-existing', + userId: 'user-1', + aggregationKey: 'key-1', + count: 2, + sent: false, + lastNotificationAt: new Date(Date.now() - 1000), // 1 second ago + cooldownSeconds: 300, // 5 minutes + }; + mockRepo.findOne.mockResolvedValue(existing); + mockRepo.save.mockResolvedValue({ ...existing, count: 3 }); + + const result = await service.checkAggregation('user-1', 'key-1'); + + expect(result.shouldSuppress).toBe(true); + expect(result.currentCount).toBe(3); + expect(existing.count).toBe(3); + }); + + it('should allow through when cooldown has expired', async () => { + const expired = { + id: 'agg-expired', + userId: 'user-1', + aggregationKey: 'key-1', + count: 5, + sent: false, + lastNotificationAt: new Date(Date.now() - 600000), // 10 minutes ago + cooldownSeconds: 300, // 5 minutes + }; + mockRepo.findOne.mockResolvedValue(expired); + + const newAgg = { + id: 'agg-new', + userId: 'user-1', + aggregationKey: 'key-1', + count: 1, + sent: false, + }; + mockRepo.create.mockReturnValue(newAgg); + mockRepo.save.mockResolvedValue(newAgg); + + const result = await service.checkAggregation('user-1', 'key-1'); + + expect(result.shouldSuppress).toBe(false); + expect(result.currentCount).toBe(1); + // Old aggregation should be marked as sent + expect(expired.sent).toBe(true); + }); + + it('should pass through without aggregation when no key provided', async () => { + const result = await service.checkAggregation('user-1', ''); + + expect(result.shouldSuppress).toBe(false); + expect(result.currentCount).toBe(0); + expect(mockRepo.findOne).not.toHaveBeenCalled(); + }); + }); + + describe('linkNotification', () => { + it('should update the aggregation with notification ID', async () => { + await service.linkNotification('agg-1', 'notif-1'); + + expect(mockRepo.update).toHaveBeenCalledWith('agg-1', { + latestNotificationId: 'notif-1', + }); + }); + }); + + describe('getAggregationStats', () => { + it('should return stats for active aggregation', async () => { + const agg = { + count: 3, + lastNotificationAt: new Date(), + }; + mockRepo.findOne.mockResolvedValue(agg); + + const stats = await service.getAggregationStats('user-1', 'key-1'); + + expect(stats).toEqual({ + count: 3, + lastNotificationAt: agg.lastNotificationAt, + }); + }); + + it('should return null when no active aggregation', async () => { + mockRepo.findOne.mockResolvedValue(null); + + const stats = await service.getAggregationStats('user-1', 'key-1'); + + expect(stats).toBeNull(); + }); + }); + + describe('cleanupOldAggregations', () => { + it('should delete old aggregation records', async () => { + const qb = { + delete: jest.fn().mockReturnThis(), + where: jest.fn().mockReturnThis(), + execute: jest.fn().mockResolvedValue({ affected: 5 }), + }; + mockRepo.createQueryBuilder.mockReturnValue(qb); + + const result = await service.cleanupOldAggregations(); + + expect(result).toBe(5); + }); + }); +}); diff --git a/src/notifications/test/notification.service.spec.ts b/src/notifications/test/notification.service.spec.ts new file mode 100644 index 0000000..f176547 --- /dev/null +++ b/src/notifications/test/notification.service.spec.ts @@ -0,0 +1,452 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { EventEmitter2 } from '@nestjs/event-emitter'; +import { NotificationService } from '../services/notification.service'; +import { NotificationTemplateService } from '../services/notification-template.service'; +import { NotificationAggregationService } from '../services/notification-aggregation.service'; +import { NotificationQueueService } from '../services/notification-queue.service'; +import { + Notification, + NotificationStatus, + NotificationChannel, + NotificationPriority, + NotificationCategory, +} from '../entities/notification.entity'; +import { NotificationDeliveryLog } from '../entities/notification-delivery-log.entity'; +import { SendNotificationDto } from '../dto/send-notification.dto'; +import { QueryNotificationHistoryDto } from '../dto/query-notification.dto'; + +describe('NotificationService', () => { + let service: NotificationService; + let notificationRepo: Repository; + let deliveryLogRepo: Repository; + let templateService: NotificationTemplateService; + let aggregationService: NotificationAggregationService; + let queueService: NotificationQueueService; + let eventEmitter: EventEmitter2; + + const mockQueryBuilder = { + select: jest.fn().mockReturnThis(), + addSelect: jest.fn().mockReturnThis(), + where: jest.fn().mockReturnThis(), + andWhere: jest.fn().mockReturnThis(), + orderBy: jest.fn().mockReturnThis(), + addOrderBy: jest.fn().mockReturnThis(), + take: jest.fn().mockReturnThis(), + getMany: jest.fn().mockResolvedValue([]), + getCount: jest.fn().mockResolvedValue(0), + update: jest.fn().mockReturnThis(), + set: jest.fn().mockReturnThis(), + execute: jest.fn().mockResolvedValue({ affected: 1 }), + groupBy: jest.fn().mockReturnThis(), + getRawMany: jest.fn().mockResolvedValue([]), + }; + + const mockNotificationRepo = { + create: jest.fn(), + save: jest.fn(), + findOne: jest.fn(), + find: jest.fn(), + update: jest.fn(), + createQueryBuilder: jest.fn(() => mockQueryBuilder), + }; + + const mockDeliveryLogRepo = { + create: jest.fn(), + save: jest.fn(), + find: jest.fn(), + update: jest.fn(), + }; + + const mockTemplateService = { + getTemplate: jest.fn(), + renderBody: jest.fn(), + }; + + const mockAggregationService = { + checkAggregation: jest.fn().mockResolvedValue({ + shouldSuppress: false, + aggregation: { id: 'agg-1', count: 1 }, + currentCount: 1, + }), + linkNotification: jest.fn(), + getAggregationStats: jest.fn(), + }; + + const mockQueueService = { + enqueueNotification: jest.fn(), + enqueueScheduledNotification: jest.fn(), + cancelNotification: jest.fn(), + }; + + const mockEventEmitter = { + emit: jest.fn(), + }; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + NotificationService, + { provide: getRepositoryToken(Notification), useValue: mockNotificationRepo }, + { provide: getRepositoryToken(NotificationDeliveryLog), useValue: mockDeliveryLogRepo }, + { provide: NotificationTemplateService, useValue: mockTemplateService }, + { provide: NotificationAggregationService, useValue: mockAggregationService }, + { provide: NotificationQueueService, useValue: mockQueueService }, + { provide: EventEmitter2, useValue: mockEventEmitter }, + ], + }).compile(); + + service = module.get(NotificationService); + notificationRepo = module.get(getRepositoryToken(Notification)); + deliveryLogRepo = module.get(getRepositoryToken(NotificationDeliveryLog)); + templateService = module.get(NotificationTemplateService); + aggregationService = module.get(NotificationAggregationService); + queueService = module.get(NotificationQueueService); + eventEmitter = module.get(EventEmitter2); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + describe('send', () => { + it('should create and queue a notification', async () => { + const dto: SendNotificationDto = { + userId: 'user-1', + title: 'Test Title', + body: 'Test Body', + category: NotificationCategory.SYSTEM, + priority: NotificationPriority.NORMAL, + primaryChannel: NotificationChannel.IN_APP, + }; + + const createdNotification = { + id: 'notif-1', + ...dto, + status: NotificationStatus.QUEUED, + read: false, + createdAt: new Date(), + aggregationCount: 0, + maxAttempts: 5, + attemptCount: 0, + deleted: false, + }; + + mockNotificationRepo.create.mockReturnValue(createdNotification); + mockNotificationRepo.save.mockResolvedValue(createdNotification); + + const result = await service.send(dto); + + expect(result).toEqual(createdNotification); + expect(mockNotificationRepo.create).toHaveBeenCalled(); + expect(mockNotificationRepo.save).toHaveBeenCalled(); + expect(mockQueueService.enqueueNotification).toHaveBeenCalledWith( + createdNotification.id, + expect.any(Object), + ); + }); + + it('should schedule a notification when scheduledAt is provided', async () => { + const futureDate = new Date(Date.now() + 3600000); + const dto: SendNotificationDto = { + userId: 'user-1', + title: 'Scheduled', + body: 'Scheduled body', + scheduledAt: futureDate.toISOString(), + }; + + const createdNotification = { + id: 'notif-sched', + ...dto, + status: NotificationStatus.SCHEDULED, + read: false, + createdAt: new Date(), + aggregationCount: 0, + maxAttempts: 5, + attemptCount: 0, + deleted: false, + }; + + mockNotificationRepo.create.mockReturnValue(createdNotification); + mockNotificationRepo.save.mockResolvedValue(createdNotification); + + await service.send(dto); + + expect(mockQueueService.enqueueScheduledNotification).toHaveBeenCalledWith( + createdNotification.id, + expect.any(Date), + ); + }); + + it('should suppress notification when aggregation triggers suppression', async () => { + mockAggregationService.checkAggregation.mockResolvedValue({ + shouldSuppress: true, + aggregation: { id: 'agg-1', count: 3 }, + currentCount: 3, + }); + mockAggregationService.getAggregationStats.mockResolvedValue(null); + + const dto: SendNotificationDto = { + userId: 'user-1', + title: 'Duplicate', + body: 'Dup body', + aggregationKey: 'portfolio-drop', + }; + + const suppressedNotif = { + id: 'notif-suppressed', + ...dto, + status: NotificationStatus.CANCELLED, + }; + + mockNotificationRepo.create.mockReturnValue(suppressedNotif); + mockNotificationRepo.save.mockResolvedValue(suppressedNotif); + + const result = await service.send(dto); + + expect(result.status).toBe(NotificationStatus.CANCELLED); + expect(mockQueueService.enqueueNotification).not.toHaveBeenCalled(); + }); + + it('should render template when templateName and templateVars are provided', async () => { + mockTemplateService.getTemplate.mockResolvedValue({ + name: 'welcome', + bodyTemplate: 'Hello {{name}}!', + htmlTemplate: '

Hello {{name}}!

', + subject: 'Welcome {{name}}', + }); + mockTemplateService.renderBody + .mockReturnValueOnce('Hello Alice!') + .mockReturnValueOnce('

Hello Alice!

') + .mockReturnValueOnce('Welcome Alice'); + + const dto: SendNotificationDto = { + userId: 'user-1', + title: 'Welcome', + body: 'Hello {{name}}!', + templateName: 'welcome', + templateVars: { name: 'Alice' }, + }; + + const createdNotif = { id: 'notif-tpl', ...dto }; + mockNotificationRepo.create.mockReturnValue(createdNotif); + mockNotificationRepo.save.mockResolvedValue(createdNotif); + + await service.send(dto); + + expect(mockTemplateService.getTemplate).toHaveBeenCalledWith('welcome'); + expect(mockTemplateService.renderBody).toHaveBeenCalledTimes(3); + }); + }); + + describe('sendBulk', () => { + it('should send multiple notifications', async () => { + const dto = { + notifications: [ + { userId: 'user-1', title: 'A', body: 'Body A' }, + { userId: 'user-2', title: 'B', body: 'Body B' }, + ], + }; + + const created = [ + { id: 'n1', userId: 'user-1', title: 'A' }, + { id: 'n2', userId: 'user-2', title: 'B' }, + ]; + + let callCount = 0; + mockNotificationRepo.create.mockImplementation(() => created[callCount++]); + mockNotificationRepo.save.mockImplementation(async (n) => n); + + const results = await service.sendBulk(dto as any); + + expect(results).toHaveLength(2); + expect(mockQueueService.enqueueNotification).toHaveBeenCalledTimes(2); + }); + }); + + describe('markAsRead', () => { + it('should mark notifications as read', async () => { + mockQueryBuilder.execute.mockResolvedValue({ affected: 3 }); + + const result = await service.markAsRead(['n1', 'n2', 'n3']); + + expect(result.updated).toBe(3); + }); + }); + + describe('markAllAsRead', () => { + it('should mark all user notifications as read', async () => { + mockQueryBuilder.execute.mockResolvedValue({ affected: 10 }); + + const result = await service.markAllAsRead('user-1'); + + expect(result.updated).toBe(10); + }); + + it('should filter by category when specified', async () => { + mockQueryBuilder.execute.mockResolvedValue({ affected: 5 }); + + await service.markAllAsRead('user-1', { category: 'security' }); + + expect(mockQueryBuilder.andWhere).toHaveBeenCalledWith( + 'category = :category', + { category: 'security' }, + ); + }); + }); + + describe('markAsUnread', () => { + it('should mark notifications as unread', async () => { + mockQueryBuilder.execute.mockResolvedValue({ affected: 2 }); + + const result = await service.markAsUnread(['n1', 'n2']); + + expect(result.updated).toBe(2); + }); + }); + + describe('getUnreadCount', () => { + it('should return unread count by category', async () => { + mockQueryBuilder.getRawMany.mockResolvedValue([ + { category: 'security', count: '3' }, + { category: 'transaction', count: '5' }, + ]); + + const result = await service.getUnreadCount('user-1'); + + expect(result.total).toBe(8); + expect(result.byCategory).toEqual({ + security: 3, + transaction: 5, + }); + }); + + it('should return zero when no unread', async () => { + mockQueryBuilder.getRawMany.mockResolvedValue([]); + + const result = await service.getUnreadCount('user-1'); + + expect(result.total).toBe(0); + expect(result.byCategory).toEqual({}); + }); + }); + + describe('getHistory', () => { + it('should return paginated notification history', async () => { + const notifications = [ + { id: 'n1', title: 'A', createdAt: new Date() }, + { id: 'n2', title: 'B', createdAt: new Date() }, + ]; + mockQueryBuilder.getMany.mockResolvedValue([...notifications]); + mockQueryBuilder.getCount.mockResolvedValue(2); + + const dto: QueryNotificationHistoryDto = { + userId: 'user-1', + limit: 10, + }; + + const result = await service.getHistory(dto); + + expect(result.data).toHaveLength(2); + expect(result.total).toBe(2); + expect(result.nextCursor).toBeNull(); + }); + + it('should generate cursor when there are more results', async () => { + const notifications = Array.from({ length: 21 }, (_, i) => ({ + id: `n${i}`, + title: `Title ${i}`, + createdAt: new Date(2025, 0, 1), + })); + mockQueryBuilder.getMany.mockResolvedValue(notifications); + + const dto: QueryNotificationHistoryDto = { + userId: 'user-1', + limit: 20, + }; + + const result = await service.getHistory(dto); + + expect(result.data).toHaveLength(20); + expect(result.nextCursor).toBeDefined(); + // Cursor should be a valid base64 string + expect(() => Buffer.from(result.nextCursor!, 'base64')).not.toThrow(); + }); + + it('should apply category filter', async () => { + mockQueryBuilder.getMany.mockResolvedValue([]); + + await service.getHistory({ + userId: 'user-1', + category: NotificationCategory.SECURITY, + }); + + expect(mockQueryBuilder.andWhere).toHaveBeenCalledWith( + 'n.category = :category', + { category: 'security' }, + ); + }); + + it('should apply read filter', async () => { + mockQueryBuilder.getMany.mockResolvedValue([]); + + await service.getHistory({ + userId: 'user-1', + read: false, + }); + + expect(mockQueryBuilder.andWhere).toHaveBeenCalledWith('n.read = false'); + }); + }); + + describe('cancel', () => { + it('should cancel a scheduled notification', async () => { + const notif = { + id: 'n1', + status: NotificationStatus.SCHEDULED, + }; + mockNotificationRepo.findOne.mockResolvedValue(notif); + mockNotificationRepo.save.mockImplementation(async (n) => n); + + const result = await service.cancel('n1'); + + expect(result.status).toBe(NotificationStatus.CANCELLED); + expect(mockQueueService.cancelNotification).toHaveBeenCalledWith('n1'); + }); + + it('should throw if notification is not scheduled', async () => { + mockNotificationRepo.findOne.mockResolvedValue({ + id: 'n1', + status: NotificationStatus.DELIVERED, + }); + + await expect(service.cancel('n1')).rejects.toThrow(); + }); + }); + + describe('trackClick', () => { + it('should mark notification as clicked', async () => { + await service.trackClick('n1'); + + expect(mockNotificationRepo.update).toHaveBeenCalledWith('n1', { + clicked: true, + clickedAt: expect.any(Date), + }); + }); + }); + + describe('softDelete', () => { + it('should soft-delete notifications', async () => { + mockQueryBuilder.execute.mockResolvedValue({ affected: 2 }); + + const result = await service.softDelete(['n1', 'n2']); + + expect(result.deleted).toBe(2); + }); + }); +}); diff --git a/src/notifications/websocket/notification.gateway.ts b/src/notifications/websocket/notification.gateway.ts new file mode 100644 index 0000000..654f366 --- /dev/null +++ b/src/notifications/websocket/notification.gateway.ts @@ -0,0 +1,229 @@ +import { + WebSocketGateway, + WebSocketServer, + OnGatewayInit, + OnGatewayConnection, + OnGatewayDisconnect, + SubscribeMessage, + MessageBody, + ConnectedSocket, + WsException, +} from '@nestjs/websockets'; +import { Server, Socket } from 'socket.io'; +import { Logger, UseFilters, UsePipes, ValidationPipe } from '@nestjs/common'; +import { OnEvent } from '@nestjs/event-emitter'; +import { JwtService } from '@nestjs/jwt'; +import { Notification } from '../entities/notification.entity'; + +/** + * Real-time notification gateway. + * + * Clients connect to the `/notifications` namespace with a valid JWT token. + * They automatically join a room named `user:` so the server can + * push notifications to a specific user. + * + * Events: + * → Server → Client: + * `notification.new` – a new notification arrived + * `notification.read` – notification was marked read + * `notification.unread_count` – updated unread count + * + * → Client → Server: + * `notification.mark_read` – mark notification(s) as read + * `notification.mark_unread` – mark notification(s) as unread + * `notification.ping` – heartbeat + */ +@WebSocketGateway({ + namespace: '/notifications', + cors: { origin: '*', credentials: true }, + pingInterval: 30000, + pingTimeout: 5000, +}) +@UsePipes(new ValidationPipe({ transform: true, whitelist: true })) +export class NotificationGateway + implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect +{ + @WebSocketServer() + server: Server; + + private readonly logger = new Logger(NotificationGateway.name); + + /** userId → Set */ + private userConnections = new Map>(); + + constructor(private readonly jwtService: JwtService) {} + + afterInit(_server: Server) { + this.logger.log('Notification WebSocket Gateway initialized'); + } + + async handleConnection(client: Socket) { + try { + const token = this.extractToken(client); + if (!token) { + client.emit('error', { code: 'AUTH_REQUIRED', message: 'Authentication token required' }); + client.disconnect(true); + return; + } + + const payload = this.jwtService.verify(token); + const userId = payload.sub || payload.userId; + if (!userId) { + throw new Error('Invalid token: missing user ID'); + } + + // Store connection + if (!this.userConnections.has(userId)) { + this.userConnections.set(userId, new Set()); + } + this.userConnections.get(userId)!.add(client.id); + + // Join user room + client.join(`user:${userId}`); + (client as any).userId = userId; + + this.logger.log(`Client connected to notifications: ${client.id} (user: ${userId})`); + + // Send connection acknowledgment + client.emit('connected', { + clientId: client.id, + timestamp: new Date().toISOString(), + heartbeatInterval: 30000, + }); + } catch (error) { + this.logger.error(`Connection error: ${error.message}`); + client.emit('error', { code: 'AUTH_FAILED', message: 'Authentication failed' }); + client.disconnect(true); + } + } + + async handleDisconnect(client: Socket) { + const userId = (client as any).userId; + if (userId) { + const connections = this.userConnections.get(userId); + if (connections) { + connections.delete(client.id); + if (connections.size === 0) { + this.userConnections.delete(userId); + } + } + this.logger.log(`Client disconnected from notifications: ${client.id} (user: ${userId})`); + } + } + + @SubscribeMessage('notification.mark_read') + async handleMarkRead( + @ConnectedSocket() client: Socket, + @MessageBody() data: { notificationIds: string[] }, + ) { + const userId = (client as any).userId; + if (!userId) throw new WsException('Not authenticated'); + + // Emit read event back to the user's room + this.server.to(`user:${userId}`).emit('notification.read', { + notificationIds: data.notificationIds, + timestamp: new Date().toISOString(), + }); + + return { event: 'ack', data: { processed: true } }; + } + + @SubscribeMessage('notification.mark_unread') + async handleMarkUnread( + @ConnectedSocket() client: Socket, + @MessageBody() data: { notificationIds: string[] }, + ) { + const userId = (client as any).userId; + if (!userId) throw new WsException('Not authenticated'); + + this.server.to(`user:${userId}`).emit('notification.unread', { + notificationIds: data.notificationIds, + timestamp: new Date().toISOString(), + }); + + return { event: 'ack', data: { processed: true } }; + } + + @SubscribeMessage('notification.ping') + async handlePing(@ConnectedSocket() client: Socket) { + return { + event: 'notification.pong', + data: { timestamp: new Date().toISOString(), serverTime: Date.now() }, + }; + } + + /** + * Listen for new notifications from the NotificationService and push + * them to the appropriate user's connected clients. + */ + @OnEvent('notification.new') + handleNewNotification(payload: { userId: string; notification: Notification }) { + this.sendToUser(payload.userId, 'notification.new', { + id: payload.notification.id, + title: payload.notification.title, + body: payload.notification.body, + category: payload.notification.category, + priority: payload.notification.priority, + primaryChannel: payload.notification.primaryChannel, + referenceId: payload.notification.referenceId, + referenceType: payload.notification.referenceType, + createdAt: payload.notification.createdAt, + }); + } + + @OnEvent('notification.read') + handleReadEvent(payload: { userId: string; notificationId: string }) { + this.sendToUser(payload.userId, 'notification.read', { + notificationId: payload.notificationId, + }); + } + + @OnEvent('notification.unread_count') + handleUnreadCountUpdate(payload: { userId: string; total: number; byCategory: Record }) { + this.sendToUser(payload.userId, 'notification.unread_count', payload); + } + + /** + * Send a message to all connected clients of a user. + */ + sendToUser(userId: string, event: string, data: any) { + this.server.to(`user:${userId}`).emit(event, data); + } + + /** + * Broadcast to all connected clients. + */ + broadcast(event: string, data: any) { + this.server.emit(event, data); + } + + /** + * Get the number of connected clients for a user. + */ + getUserConnectionCount(userId: string): number { + return this.userConnections.get(userId)?.size || 0; + } + + /** + * Get total connected client count. + */ + getTotalConnectionCount(): number { + let count = 0; + for (const connections of this.userConnections.values()) { + count += connections.size; + } + return count; + } + + private extractToken(client: Socket): string | null { + const authHeader = client.handshake.headers.authorization; + if (authHeader?.startsWith('Bearer ')) { + return authHeader.substring(7); + } + const tokenAuth = client.handshake.auth?.token; + if (tokenAuth) return tokenAuth; + const tokenQuery = client.handshake.query?.token; + if (typeof tokenQuery === 'string') return tokenQuery; + return null; + } +}