feat: implement notification system with real-time stream updates and automated financial alerts

.agents/memory.md

@@ -51,6 +51,9 @@ File này chỉ lưu sự thật và quyết định dài hạn giúp các phiê
 - Notification dùng inbox theo ownership và database-backed delivery outbox với khóa chống
   trùng theo sự kiện. Kênh mặc định là `IN_APP`; email dùng SMTP hiện có, còn Zalo/push giữ
   trạng thái delivery riêng để bổ sung provider adapter sau.
+  Thông báo in-app và số lượng chưa đọc (`unread-count`) được phát thời gian thực tới client qua
+  Server-Sent Events (`GET /api/v1/notifications/stream`), quản lý kết nối và phát sóng bởi
+  `NotificationStreamService` (hỗ trợ Redis Pub/Sub đa instance và in-memory fallback).
 - Reminder hỗ trợ `ONCE`, `DAILY`, `WEEKLY`, `MONTHLY`, `YEARLY`, có khoảng lặp và ngày kết
   thúc. Worker nền trong process xử lý reminder, cảnh báo ngân sách/mục tiêu và retry delivery;
   có thể tắt hoặc chỉnh chu kỳ bằng các biến `NOTIFICATION_*`.

src/config/swagger.config.ts

@@ -4893,6 +4893,36 @@ export const swaggerSpec = {
         },
       },
     },
+    '/notifications/stream': {
+      get: {
+        tags: ['Notifications'],
+        summary: 'Stream realtime notifications and unread count via Server-Sent Events (SSE)',
+        security: [{ BearerAuth: [] }],
+        parameters: [
+          {
+            name: 'token',
+            in: 'query',
+            description: 'Access token for EventSource authentication when custom headers are not supported',
+            required: false,
+            schema: { type: 'string' },
+          },
+        ],
+        responses: {
+          200: {
+            description: 'Server-Sent Events stream emitting "connected", "unread_count", and "notification" events',
+            content: {
+              'text/event-stream': {
+                schema: {
+                  type: 'string',
+                  example: 'event: unread_count\ndata: {"count": 3}\n\n',
+                },
+              },
+            },
+          },
+          401: { $ref: '#/components/responses/Unauthorized' },
+        },
+      },
+    },
     '/notifications/unread-count': {
       get: {
         tags: ['Notifications'],

src/middlewares/auth.middleware.ts

@@ -20,6 +20,10 @@ export async function authMiddleware(
     }
   }
 
+  if (!token && typeof req.query?.token === 'string') {
+    token = req.query.token;
+  }
+
   if (!token) {
     next(new AppError('Unauthorized', 401, ERROR_CODE.UNAUTHORIZED));
     return;

src/modules/notifications/notification-stream.service.ts

+import { Request, Response } from 'express';
+import Redis from 'ioredis';
+import { envConfig } from '../../config/env.config';
+import { LoggerService } from '../../common/services/logger.service';
+import { NotificationRepository } from './notification.repository';
+
+interface SsePayload {
+  userId: string;
+  event: string;
+  data: any;
+}
+
+export class NotificationStreamService {
+  private readonly logger = new LoggerService('NotificationStreamService');
+  private readonly repository = new NotificationRepository();
+  private readonly clients = new Map<string, Set<Response>>();
+
+  private publisher: Redis | null = null;
+  private subscriber: Redis | null = null;
+  private heartbeatInterval: NodeJS.Timeout | null = null;
+  private isRedisPubSubReady = false;
+
+  constructor() {
+    this.initPubSub();
+    this.initHeartbeat();
+  }
+
+  private initPubSub() {
+    if (!envConfig.redis.enabled) {
+      this.logger.info('Redis disabled, notification stream running in standalone in-memory mode.');
+      return;
+    }
+
+    try {
+      const retryStrategy = (times: number) => {
+        if (times > 3) {
+          this.logger.warn('Failed to connect Redis for SSE Pub/Sub. Falling back to in-memory mode.');
+          this.isRedisPubSubReady = false;
+          return null;
+        }
+        return Math.min(times * 100, 2000);
+      };
+
+      const redisOptions = envConfig.redis.url
+        ? { lazyConnect: true, retryStrategy }
+        : {
+            host: envConfig.redis.host,
+            port: envConfig.redis.port,
+            password: envConfig.redis.password,
+            lazyConnect: true,
+            retryStrategy,
+          };
+
+      this.publisher = envConfig.redis.url
+        ? new Redis(envConfig.redis.url, redisOptions)
+        : new Redis(redisOptions);
+
+      this.subscriber = envConfig.redis.url
+        ? new Redis(envConfig.redis.url, redisOptions)
+        : new Redis(redisOptions);
+
+      const channel = 'finwise:notifications:sse';
+
+      this.subscriber.subscribe(channel, (err) => {
+        if (err) {
+          this.logger.error('Failed to subscribe to Redis SSE channel:', err);
+          this.isRedisPubSubReady = false;
+        } else {
+          this.isRedisPubSubReady = true;
+          this.logger.info(`Subscribed to Redis SSE channel "${channel}".`);
+        }
+      });
+
+      this.subscriber.on('message', (_channel, message) => {
+        try {
+          const payload = JSON.parse(message) as SsePayload;
+          if (payload?.userId && payload?.event) {
+            this.sendToLocalClients(payload.userId, payload.event, payload.data);
+          }
+        } catch (error) {
+          this.logger.error('Error processing Redis SSE message:', error);
+        }
+      });
+
+      this.subscriber.on('error', (err) => {
+        this.logger.warn('Redis SSE subscriber error:', err.message);
+        this.isRedisPubSubReady = false;
+      });
+
+      this.publisher.on('error', (err) => {
+        this.logger.warn('Redis SSE publisher error:', err.message);
+        this.isRedisPubSubReady = false;
+      });
+    } catch (error) {
+      this.logger.warn('Error setting up Redis SSE pub/sub:', error);
+      this.isRedisPubSubReady = false;
+    }
+  }
+
+  private initHeartbeat() {
+    // Send comment ping every 25 seconds to keep connections alive through proxies
+    this.heartbeatInterval = setInterval(() => {
+      this.pingAll();
+    }, 25_000);
+    this.heartbeatInterval.unref();
+  }
+
+  private pingAll() {
+    for (const [userId, userClients] of this.clients.entries()) {
+      for (const client of userClients) {
+        try {
+          if (!client.writableEnded && !client.destroyed) {
+            client.write(': ping\n\n');
+          } else {
+            userClients.delete(client);
+          }
+        } catch {
+          userClients.delete(client);
+        }
+      }
+      if (userClients.size === 0) {
+        this.clients.delete(userId);
+      }
+    }
+  }
+
+  async registerClient(userId: string, res: Response, req: Request): Promise<void> {
+    // Register response connection
+    if (!this.clients.has(userId)) {
+      this.clients.set(userId, new Set<Response>());
+    }
+    const userClients = this.clients.get(userId)!;
+    userClients.add(res);
+
+    this.logger.info(`SSE client connected for user ${userId} (active connections: ${userClients.size})`);
+
+    // Clean up when client disconnects
+    const cleanup = () => {
+      const current = this.clients.get(userId);
+      if (current) {
+        current.delete(res);
+        if (current.size === 0) {
+          this.clients.delete(userId);
+        }
+      }
+      this.logger.info(`SSE client disconnected for user ${userId}`);
+    };
+
+    if (typeof req.on === 'function') {
+      req.on('close', cleanup);
+    }
+    if (typeof res.on === 'function') {
+      res.on('error', cleanup);
+    }
+
+    // 1. Send handshake connected event
+    this.writeEvent(res, 'connected', {
+      connectedAt: new Date().toISOString(),
+      userId,
+    });
+
+    // 2. Send current unread count immediately so client has latest state
+    try {
+      const count = await this.repository.unreadCount(userId);
+      this.writeEvent(res, 'unread_count', { count });
+    } catch (error) {
+      this.logger.error(`Failed to send initial unread count to user ${userId}:`, error);
+    }
+  }
+
+  broadcastToUser(userId: string, event: string, data: any): void {
+    if (this.isRedisPubSubReady && this.publisher) {
+      const payload: SsePayload = { userId, event, data };
+      this.publisher.publish('finwise:notifications:sse', JSON.stringify(payload)).catch((err) => {
+        this.logger.warn('Failed to publish SSE event to Redis, falling back to local dispatch:', err);
+        this.sendToLocalClients(userId, event, data);
+      });
+    } else {
+      this.sendToLocalClients(userId, event, data);
+    }
+  }
+
+  private sendToLocalClients(userId: string, event: string, data: any): void {
+    const userClients = this.clients.get(userId);
+    if (!userClients || userClients.size === 0) {
+      return;
+    }
+
+    for (const client of userClients) {
+      try {
+        if (!client.writableEnded && !client.destroyed) {
+          this.writeEvent(client, event, data);
+        } else {
+          userClients.delete(client);
+        }
+      } catch (err) {
+        this.logger.warn(`Error writing SSE event to client of user ${userId}:`, err);
+        userClients.delete(client);
+      }
+    }
+
+    if (userClients.size === 0) {
+      this.clients.delete(userId);
+    }
+  }
+
+  private writeEvent(res: Response, event: string, data: any): void {
+    res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
+    if (typeof (res as any).flush === 'function') {
+      (res as any).flush();
+    }
+  }
+
+  getActiveConnectionCount(userId?: string): number {
+    if (userId) {
+      return this.clients.get(userId)?.size || 0;
+    }
+    let total = 0;
+    for (const set of this.clients.values()) {
+      total += set.size;
+    }
+    return total;
+  }
+
+  shutdown(): void {
+    if (this.heartbeatInterval) {
+      clearInterval(this.heartbeatInterval);
+      this.heartbeatInterval = null;
+    }
+
+    for (const [userId, userClients] of this.clients.entries()) {
+      for (const client of userClients) {
+        try {
+          client.end();
+        } catch {
+          // ignore error on close
+        }
+      }
+    }
+    this.clients.clear();
+
+    if (this.subscriber) {
+      this.subscriber.disconnect();
+      this.subscriber = null;
+    }
+    if (this.publisher) {
+      this.publisher.disconnect();
+      this.publisher = null;
+    }
+  }
+}
+
+export const notificationStreamService = new NotificationStreamService();

src/modules/notifications/notification.controller.ts

@@ -4,10 +4,28 @@ import {
   UpdateNotificationSettingDto,
 } from './notification.dto';
 import { NotificationService } from './notification.service';
+import { notificationStreamService } from './notification-stream.service';
 
 export class NotificationController {
   private readonly service = new NotificationService();
 
+  stream = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      res.writeHead(200, {
+        'Content-Type': 'text/event-stream',
+        'Cache-Control': 'no-cache, no-transform',
+        'Connection': 'keep-alive',
+        'X-Accel-Buffering': 'no',
+      });
+      if (typeof res.flushHeaders === 'function') {
+        res.flushHeaders();
+      }
+      await notificationStreamService.registerClient(req.user.id, res, req);
+    } catch (error) {
+      next(error);
+    }
+  };
+
   findAll = async (req: Request, res: Response, next: NextFunction) => {
     try {
       const result = await this.service.findAll(

src/modules/notifications/notification.route.ts

@@ -19,6 +19,7 @@ const controller = new NotificationController();
 router.use(authMiddleware);
 
 router.get('/', requirePermission(PERMISSIONS.NOTIFICATION_READ), validate(findNotificationsSchema, 'query'), controller.findAll);
+router.get('/stream', requirePermission(PERMISSIONS.NOTIFICATION_READ), controller.stream);
 router.get('/unread-count', requirePermission(PERMISSIONS.NOTIFICATION_READ), controller.unreadCount);
 router.patch('/read-all', requirePermission(PERMISSIONS.NOTIFICATION_UPDATE), controller.markAllRead);
 router.get('/settings', requirePermission(PERMISSIONS.NOTIFICATION_READ), controller.getSetting);

src/modules/notifications/notification.service.ts

@@ -29,6 +29,7 @@ import {
 
 import { AnomalyService } from '../anomalies/anomaly.service';
 import { webhookService } from '../webhooks/webhook.service';
+import { notificationStreamService } from './notification-stream.service';
 
 const GOAL_NEAR_TARGET_PERCENT = new Prisma.Decimal(80);
 const GOAL_DUE_SOON_DAYS = 7;
@@ -57,18 +58,43 @@ export class NotificationService {
 
   async markRead(userId: string, id: string) {
     const notification = await this.findOwned(userId, id);
-    return notification.readAt
+    const updated = notification.readAt
       ? notification
-      : this.repository.markRead(notification.id);
+      : await this.repository.markRead(notification.id);
+
+    if (!notification.readAt) {
+      try {
+        const count = await this.repository.unreadCount(userId);
+        notificationStreamService.broadcastToUser(userId, 'unread_count', { count });
+      } catch (err) {
+        console.error('Failed to broadcast unread_count on markRead:', err);
+      }
+    }
+    return updated;
   }
 
   async markAllRead(userId: string) {
-    return this.repository.markAllRead(userId);
+    const result = await this.repository.markAllRead(userId);
+    try {
+      notificationStreamService.broadcastToUser(userId, 'unread_count', { count: 0 });
+    } catch (err) {
+      console.error('Failed to broadcast unread_count on markAllRead:', err);
+    }
+    return result;
   }
 
   async remove(userId: string, id: string) {
-    await this.findOwned(userId, id);
-    return this.repository.remove(id);
+    const notification = await this.findOwned(userId, id);
+    const result = await this.repository.remove(id);
+    if (!notification.readAt) {
+      try {
+        const count = await this.repository.unreadCount(userId);
+        notificationStreamService.broadcastToUser(userId, 'unread_count', { count });
+      } catch (err) {
+        console.error('Failed to broadcast unread_count on remove:', err);
+      }
+    }
+    return result;
   }
 
   async getSetting(userId: string): Promise<NotificationSettingDto> {
@@ -85,7 +111,17 @@ export class NotificationService {
     if (!this.isEnabled(input.type, setting)) {
       return null;
     }
-    return this.repository.createIfAbsent(input, setting.channels);
+    const created = await this.repository.createIfAbsent(input, setting.channels);
+    if (created) {
+      try {
+        notificationStreamService.broadcastToUser(input.userId, 'notification', created);
+        const count = await this.repository.unreadCount(input.userId);
+        notificationStreamService.broadcastToUser(input.userId, 'unread_count', { count });
+      } catch (err) {
+        console.error('Failed to broadcast notification/unread_count on create:', err);
+      }
+    }
+    return created;
   }
 
   async getChannelsForType(type: NotificationType, userId: string) {

tests/notification-stream.test.ts

+import http from 'http';
+import request from 'supertest';
+import app from '../src/app';
+import jwt from 'jsonwebtoken';
+import { jwtConfig } from '../src/config/jwt.config';
+import { notificationStreamService } from '../src/modules/notifications/notification-stream.service';
+import { prisma } from '../src/database/prisma.client';
+import { SYSTEM_ROLES } from '../src/common/constants';
+
+describe('Notification SSE Stream Integration Tests', () => {
+  let userToken: string;
+  let userId: string;
+
+  beforeAll(async () => {
+    const userRole = await prisma.role.findUnique({
+      where: { name: SYSTEM_ROLES.USER },
+    });
+
+    const user = await prisma.user.create({
+      data: {
+        email: `sse-test-${Date.now()}@finwise.local`,
+        password: 'hashedpassword',
+        fullName: 'SSE Test User',
+        isActive: true,
+        roleId: userRole!.id,
+      },
+    });
+    userId = user.id;
+
+    userToken = jwt.sign(
+      { id: user.id, email: user.email, role: SYSTEM_ROLES.USER },
+      jwtConfig.accessSecret,
+      { expiresIn: '1h' },
+    );
+  });
+
+  afterAll(async () => {
+    notificationStreamService.shutdown();
+    if (userId) {
+      await prisma.notification.deleteMany({ where: { userId } });
+      await prisma.user.deleteMany({ where: { id: userId } });
+    }
+  });
+
+  it('should reject unauthorized connection with 401', async () => {
+    const res = await request(app).get('/api/v1/notifications/stream');
+    expect(res.status).toBe(401);
+  });
+
+  it('should accept connection with query token and establish text/event-stream headers', (done) => {
+    const server = http.createServer(app);
+    server.listen(0, () => {
+      const addr = server.address() as any;
+      const port = addr.port;
+      const clientReq = http.get(
+        `http://127.0.0.1:${port}/api/v1/notifications/stream?token=${encodeURIComponent(userToken)}`,
+        (res) => {
+          expect(res.statusCode).toBe(200);
+          expect(res.headers['content-type']).toContain('text/event-stream');
+          expect(res.headers['cache-control']).toContain('no-cache');
+
+          let receivedData = '';
+          res.on('data', (chunk) => {
+            receivedData += chunk.toString();
+            if (receivedData.includes('event: connected') && receivedData.includes('event: unread_count')) {
+              expect(receivedData).toContain('event: connected');
+              expect(receivedData).toContain('event: unread_count');
+              clientReq.destroy();
+              server.close(() => done());
+            }
+          });
+        },
+      );
+
+      clientReq.on('error', (err: any) => {
+        if (err.code === 'ECONNRESET' || clientReq.destroyed) {
+          return;
+        }
+        server.close(() => done(err));
+      });
+    });
+  });
+
+  it('should broadcast notification and unread count to active connections', async () => {
+    let mockChunk = '';
+    const mockRes: any = {
+      writableEnded: false,
+      destroyed: false,
+      write: jest.fn((chunk: string) => {
+        mockChunk += chunk;
+        return true;
+      }),
+      on: jest.fn(),
+      end: jest.fn(),
+    };
+    const mockReq: any = {
+      on: jest.fn(),
+    };
+
+    await notificationStreamService.registerClient(userId, mockRes, mockReq);
+    expect(notificationStreamService.getActiveConnectionCount(userId)).toBeGreaterThanOrEqual(1);
+
+    // Broadcast test event
+    notificationStreamService.broadcastToUser(userId, 'unread_count', { count: 5 });
+    expect(mockChunk).toContain('event: unread_count');
+    expect(mockChunk).toContain('"count":5');
+
+    // Simulate disconnect
+    const closeHandler = mockReq.on.mock.calls.find((c: any) => c[0] === 'close')?.[1];
+    if (closeHandler) {
+      closeHandler();
+    }
+  });
+});