feat：实现通知系统核心功能

- 添加通知实体和数据传输对象 - 实现通知服务层逻辑，支持创建、查询、标记已读 - 添加通知REST API控制器 - 实现WebSocket网关，支持实时通知推送 - 支持系统通知、用户通知、广播通知三种类型 - 支持定时通知功能，每分钟自动检查待发送通知 - 添加通知模块导出
2026-01-10 21:51:29 +08:00
parent 23c54215e1
commit a2d630d864
8 changed files with 521 additions and 0 deletions
--- a/src/business/notice/dto/create-notice.dto.ts
+++ b/src/business/notice/dto/create-notice.dto.ts
@@ -0,0 +1,38 @@
+import { IsString, IsOptional, IsNumber, IsEnum, IsDateString, IsObject } from 'class-validator';
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+import { NoticeType } from '../notice.entity';
+
+export class CreateNoticeDto {
+  @ApiProperty({ description: '通知标题' })
+  @IsString()
+  title: string;
+
+  @ApiProperty({ description: '通知内容' })
+  @IsString()
+  content: string;
+
+  @ApiPropertyOptional({ enum: NoticeType, description: '通知类型' })
+  @IsOptional()
+  @IsEnum(NoticeType)
+  type?: NoticeType;
+
+  @ApiPropertyOptional({ description: '接收者用户ID，不填表示广播' })
+  @IsOptional()
+  @IsNumber()
+  userId?: number;
+
+  @ApiPropertyOptional({ description: '发送者用户ID' })
+  @IsOptional()
+  @IsNumber()
+  senderId?: number;
+
+  @ApiPropertyOptional({ description: '计划发送时间' })
+  @IsOptional()
+  @IsDateString()
+  scheduledAt?: string;
+
+  @ApiPropertyOptional({ description: '额外元数据' })
+  @IsOptional()
+  @IsObject()
+  metadata?: Record<string, any>;
+}
--- a/src/business/notice/dto/notice-response.dto.ts
+++ b/src/business/notice/dto/notice-response.dto.ts
@@ -0,0 +1,43 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { NoticeType, NoticeStatus } from '../notice.entity';
+
+export class NoticeResponseDto {
+  @ApiProperty()
+  id: number;
+
+  @ApiProperty()
+  title: string;
+
+  @ApiProperty()
+  content: string;
+
+  @ApiProperty({ enum: NoticeType })
+  type: NoticeType;
+
+  @ApiProperty({ enum: NoticeStatus })
+  status: NoticeStatus;
+
+  @ApiProperty({ nullable: true })
+  userId: number | null;
+
+  @ApiProperty({ nullable: true })
+  senderId: number | null;
+
+  @ApiProperty({ nullable: true })
+  scheduledAt: Date | null;
+
+  @ApiProperty({ nullable: true })
+  sentAt: Date | null;
+
+  @ApiProperty({ nullable: true })
+  readAt: Date | null;
+
+  @ApiProperty({ nullable: true })
+  metadata: Record<string, any> | null;
+
+  @ApiProperty()
+  createdAt: Date;
+
+  @ApiProperty()
+  updatedAt: Date;
+}
--- a/src/business/notice/index.ts
+++ b/src/business/notice/index.ts
@@ -0,0 +1,7 @@
+export * from './notice.entity';
+export * from './notice.service';
+export * from './notice.controller';
+export * from './notice.gateway';
+export * from './notice.module';
+export * from './dto/create-notice.dto';
+export * from './dto/notice-response.dto';
--- a/src/business/notice/notice.controller.ts
+++ b/src/business/notice/notice.controller.ts
@@ -0,0 +1,87 @@
+import {
+  Controller,
+  Get,
+  Post,
+  Body,
+  Param,
+  Patch,
+  Query,
+  ParseIntPipe,
+  UseGuards,
+} from '@nestjs/common';
+import { ApiTags, ApiOperation, ApiResponse, ApiBearerAuth } from '@nestjs/swagger';
+import { NoticeService } from './notice.service';
+import { CreateNoticeDto } from './dto/create-notice.dto';
+import { NoticeResponseDto } from './dto/notice-response.dto';
+import { JwtAuthGuard } from '../auth/jwt_auth.guard';
+import { CurrentUser } from '../auth/current_user.decorator';
+
+@ApiTags('通知管理')
+@Controller('api/notices')
+@UseGuards(JwtAuthGuard)
+@ApiBearerAuth()
+export class NoticeController {
+  constructor(private readonly noticeService: NoticeService) {}
+
+  @Post()
+  @ApiOperation({ summary: '创建通知' })
+  @ApiResponse({ status: 201, description: '通知创建成功', type: NoticeResponseDto })
+  async create(@Body() createNoticeDto: CreateNoticeDto): Promise<NoticeResponseDto> {
+    return this.noticeService.create(createNoticeDto);
+  }
+
+  @Get()
+  @ApiOperation({ summary: '获取通知列表' })
+  @ApiResponse({ status: 200, description: '获取成功', type: [NoticeResponseDto] })
+  async findAll(
+    @CurrentUser() user: any,
+    @Query('all') all?: string,
+  ): Promise<NoticeResponseDto[]> {
+    // 如果是管理员且指定了all参数，返回所有通知
+    const userId = all === 'true' && user.isAdmin ? undefined : user.id;
+    return this.noticeService.findAll(userId);
+  }
+
+  @Get('unread-count')
+  @ApiOperation({ summary: '获取未读通知数量' })
+  @ApiResponse({ status: 200, description: '获取成功' })
+  async getUnreadCount(@CurrentUser() user: any): Promise<{ count: number }> {
+    const count = await this.noticeService.getUserUnreadCount(user.id);
+    return { count };
+  }
+
+  @Get(':id')
+  @ApiOperation({ summary: '获取通知详情' })
+  @ApiResponse({ status: 200, description: '获取成功', type: NoticeResponseDto })
+  async findOne(@Param('id', ParseIntPipe) id: number): Promise<NoticeResponseDto> {
+    return this.noticeService.findById(id);
+  }
+
+  @Patch(':id/read')
+  @ApiOperation({ summary: '标记通知为已读' })
+  @ApiResponse({ status: 200, description: '标记成功', type: NoticeResponseDto })
+  async markAsRead(
+    @Param('id', ParseIntPipe) id: number,
+    @CurrentUser() user: any,
+  ): Promise<NoticeResponseDto> {
+    return this.noticeService.markAsRead(id, user.id);
+  }
+
+  @Post('system')
+  @ApiOperation({ summary: '发送系统通知' })
+  @ApiResponse({ status: 201, description: '发送成功', type: NoticeResponseDto })
+  async sendSystemNotice(
+    @Body() body: { title: string; content: string; userId?: number },
+  ): Promise<NoticeResponseDto> {
+    return this.noticeService.sendSystemNotice(body.title, body.content, body.userId);
+  }
+
+  @Post('broadcast')
+  @ApiOperation({ summary: '发送广播通知' })
+  @ApiResponse({ status: 201, description: '发送成功', type: NoticeResponseDto })
+  async sendBroadcast(
+    @Body() body: { title: string; content: string },
+  ): Promise<NoticeResponseDto> {
+    return this.noticeService.sendBroadcast(body.title, body.content);
+  }
+}
--- a/src/business/notice/notice.entity.ts
+++ b/src/business/notice/notice.entity.ts
@@ -0,0 +1,64 @@
+import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn } from 'typeorm';
+
+export enum NoticeType {
+  SYSTEM = 'system',
+  USER = 'user',
+  BROADCAST = 'broadcast',
+}
+
+export enum NoticeStatus {
+  PENDING = 'pending',
+  SENT = 'sent',
+  READ = 'read',
+  FAILED = 'failed',
+}
+
+@Entity('notices')
+export class Notice {
+  @PrimaryGeneratedColumn()
+  id: number;
+
+  @Column()
+  title: string;
+
+  @Column('text')
+  content: string;
+
+  @Column({
+    type: 'enum',
+    enum: NoticeType,
+    default: NoticeType.SYSTEM,
+  })
+  type: NoticeType;
+
+  @Column({
+    type: 'enum',
+    enum: NoticeStatus,
+    default: NoticeStatus.PENDING,
+  })
+  status: NoticeStatus;
+
+  @Column({ nullable: true })
+  userId: number; // 接收者ID，null表示广播通知
+
+  @Column({ nullable: true })
+  senderId: number; // 发送者ID
+
+  @Column({ type: 'datetime', nullable: true })
+  scheduledAt: Date; // 计划发送时间
+
+  @Column({ type: 'datetime', nullable: true })
+  sentAt: Date; // 实际发送时间
+
+  @Column({ type: 'datetime', nullable: true })
+  readAt: Date; // 阅读时间
+
+  @Column({ type: 'json', nullable: true })
+  metadata: Record<string, any>; // 额外数据
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+}
--- a/src/business/notice/notice.gateway.ts
+++ b/src/business/notice/notice.gateway.ts
@@ -0,0 +1,117 @@
+import {
+  WebSocketGateway,
+  WebSocketServer,
+  SubscribeMessage,
+  MessageBody,
+  ConnectedSocket,
+  OnGatewayConnection,
+  OnGatewayDisconnect,
+} from '@nestjs/websockets';
+import { Server } from 'ws';
+import * as WebSocket from 'ws';
+import { Logger } from '@nestjs/common';
+
+interface AuthenticatedSocket extends WebSocket {
+  userId?: number;
+}
+
+@WebSocketGateway({
+  cors: {
+    origin: '*',
+  },
+  path: '/ws/notice',
+})
+export class NoticeGateway implements OnGatewayConnection, OnGatewayDisconnect {
+  @WebSocketServer()
+  server: Server;
+
+  private readonly logger = new Logger(NoticeGateway.name);
+  private readonly userSockets = new Map<number, Set<AuthenticatedSocket>>();
+
+  handleConnection(client: AuthenticatedSocket) {
+    this.logger.log(`Client connected: ${client.readyState}`);
+  }
+
+  handleDisconnect(client: AuthenticatedSocket) {
+    this.logger.log(`Client disconnected`);
+    
+    if (client.userId) {
+      const userSockets = this.userSockets.get(client.userId);
+      if (userSockets) {
+        userSockets.delete(client);
+        if (userSockets.size === 0) {
+          this.userSockets.delete(client.userId);
+        }
+      }
+    }
+  }
+
+  @SubscribeMessage('authenticate')
+  handleAuthenticate(
+    @MessageBody() data: { userId: number },
+    @ConnectedSocket() client: AuthenticatedSocket,
+  ) {
+    const { userId } = data;
+    
+    if (!userId) {
+      client.send(JSON.stringify({ error: 'User ID is required' }));
+      return;
+    }
+
+    client.userId = userId;
+    
+    if (!this.userSockets.has(userId)) {
+      this.userSockets.set(userId, new Set());
+    }
+    this.userSockets.get(userId)!.add(client);
+
+    client.send(JSON.stringify({ 
+      type: 'authenticated', 
+      data: { userId } 
+    }));
+
+    this.logger.log(`User ${userId} authenticated`);
+  }
+
+  @SubscribeMessage('ping')
+  handlePing(@ConnectedSocket() client: AuthenticatedSocket) {
+    client.send(JSON.stringify({ type: 'pong' }));
+  }
+
+  // 发送消息给特定用户
+  sendToUser(userId: number, message: any) {
+    const userSockets = this.userSockets.get(userId);
+    if (userSockets) {
+      const messageStr = JSON.stringify(message);
+      userSockets.forEach(socket => {
+        if (socket.readyState === WebSocket.OPEN) {
+          socket.send(messageStr);
+        }
+      });
+      this.logger.log(`Message sent to user ${userId}`);
+    } else {
+      this.logger.warn(`User ${userId} not connected`);
+    }
+  }
+
+  // 广播消息给所有连接的用户
+  broadcast(message: any) {
+    const messageStr = JSON.stringify(message);
+    this.server.clients.forEach(client => {
+      if (client.readyState === WebSocket.OPEN) {
+        client.send(messageStr);
+      }
+    });
+    this.logger.log('Message broadcasted to all clients');
+  }
+
+  // 获取在线用户数量
+  getOnlineUsersCount(): number {
+    return this.userSockets.size;
+  }
+
+  // 获取在线用户列表
+  getOnlineUsers(): number[] {
+    return Array.from(this.userSockets.keys());
+  }
+}
--- a/src/business/notice/notice.module.ts
+++ b/src/business/notice/notice.module.ts
@@ -0,0 +1,20 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { ScheduleModule } from '@nestjs/schedule';
+import { Notice } from './notice.entity';
+import { NoticeService } from './notice.service';
+import { NoticeController } from './notice.controller';
+import { NoticeGateway } from './notice.gateway';
+import { LoginCoreModule } from '../../core/login_core/login_core.module';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([Notice]),
+    ScheduleModule.forRoot(),
+    LoginCoreModule,
+  ],
+  controllers: [NoticeController],
+  providers: [NoticeService, NoticeGateway],
+  exports: [NoticeService, NoticeGateway],
+})
+export class NoticeModule {}
--- a/src/business/notice/notice.service.ts
+++ b/src/business/notice/notice.service.ts
@@ -0,0 +1,145 @@
+import { Injectable, Logger, NotFoundException } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository, LessThanOrEqual } from 'typeorm';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { Notice, NoticeStatus, NoticeType } from './notice.entity';
+import { CreateNoticeDto } from './dto/create-notice.dto';
+import { NoticeGateway } from './notice.gateway';
+
+@Injectable()
+export class NoticeService {
+  private readonly logger = new Logger(NoticeService.name);
+
+  constructor(
+    @InjectRepository(Notice)
+    private readonly noticeRepository: Repository<Notice>,
+    private readonly noticeGateway: NoticeGateway,
+  ) {}
+
+  async create(createNoticeDto: CreateNoticeDto): Promise<Notice> {
+    const notice = this.noticeRepository.create({
+      ...createNoticeDto,
+      scheduledAt: createNoticeDto.scheduledAt ? new Date(createNoticeDto.scheduledAt) : null,
+    });
+
+    const savedNotice = await this.noticeRepository.save(notice);
+
+    // 如果没有设置计划时间，立即发送
+    if (!savedNotice.scheduledAt) {
+      await this.sendNotice(savedNotice);
+    }
+
+    return savedNotice;
+  }
+
+  async findAll(userId?: number): Promise<Notice[]> {
+    const query = this.noticeRepository.createQueryBuilder('notice');
+    
+    if (userId) {
+      query.where('notice.userId = :userId OR notice.userId IS NULL', { userId });
+    }
+    
+    return query.orderBy('notice.createdAt', 'DESC').getMany();
+  }
+
+  async findById(id: number): Promise<Notice> {
+    const notice = await this.noticeRepository.findOne({ where: { id } });
+    if (!notice) {
+      throw new NotFoundException(`Notice with ID ${id} not found`);
+    }
+    return notice;
+  }
+
+  async markAsRead(id: number, userId?: number): Promise<Notice> {
+    const notice = await this.findById(id);
+    
+    // 检查权限：只能标记自己的通知或广播通知为已读
+    if (notice.userId && userId && notice.userId !== userId) {
+      throw new NotFoundException(`Notice with ID ${id} not found`);
+    }
+
+    notice.status = NoticeStatus.READ;
+    notice.readAt = new Date();
+    
+    return this.noticeRepository.save(notice);
+  }
+
+  async getUserUnreadCount(userId: number): Promise<number> {
+    return this.noticeRepository.count({
+      where: [
+        { userId, status: NoticeStatus.SENT },
+        { userId: null, status: NoticeStatus.SENT }, // 广播通知
+      ],
+    });
+  }
+
+  private async sendNotice(notice: Notice): Promise<void> {
+    try {
+      // 通过WebSocket发送通知
+      if (notice.userId) {
+        // 发送给特定用户
+        this.noticeGateway.sendToUser(notice.userId, {
+          type: 'notice',
+          data: notice,
+        });
+      } else {
+        // 广播通知
+        this.noticeGateway.broadcast({
+          type: 'notice',
+          data: notice,
+        });
+      }
+
+      // 更新状态
+      notice.status = NoticeStatus.SENT;
+      notice.sentAt = new Date();
+      await this.noticeRepository.save(notice);
+
+      this.logger.log(`Notice ${notice.id} sent successfully`);
+    } catch (error) {
+      this.logger.error(`Failed to send notice ${notice.id}:`, error);
+      
+      notice.status = NoticeStatus.FAILED;
+      await this.noticeRepository.save(notice);
+    }
+  }
+
+  // 定时任务：每分钟检查需要发送的通知
+  @Cron(CronExpression.EVERY_MINUTE)
+  async handleScheduledNotices(): Promise<void> {
+    const now = new Date();
+    const pendingNotices = await this.noticeRepository.find({
+      where: {
+        status: NoticeStatus.PENDING,
+        scheduledAt: LessThanOrEqual(now),
+      },
+    });
+
+    for (const notice of pendingNotices) {
+      await this.sendNotice(notice);
+    }
+
+    if (pendingNotices.length > 0) {
+      this.logger.log(`Processed ${pendingNotices.length} scheduled notices`);
+    }
+  }
+
+  // 发送系统通知的便捷方法
+  async sendSystemNotice(title: string, content: string, userId?: number): Promise<Notice> {
+    return this.create({
+      title,
+      content,
+      type: NoticeType.SYSTEM,
+      userId,
+    });
+  }
+
+  // 发送广播通知的便捷方法
+  async sendBroadcast(title: string, content: string): Promise<Notice> {
+    return this.create({
+      title,
+      content,
+      type: NoticeType.BROADCAST,
+    });
+  }
+}