101 lines
3.2 KiB
TypeScript
101 lines
3.2 KiB
TypeScript
import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
|
|
import { InjectRepository } from '@nestjs/typeorm';
|
|
import { Repository } from 'typeorm';
|
|
import { Subject, Observable } from 'rxjs';
|
|
import { EventEmitter2 } from '@nestjs/event-emitter';
|
|
import { Notification } from '../entities/notification.entity';
|
|
import { CreateNotificationDto } from './dto/notification.dto';
|
|
|
|
@Injectable()
|
|
export class NotificationsService {
|
|
private subjects = new Map<number, Subject<Notification>>();
|
|
private subscriberCounts = new Map<number, number>();
|
|
|
|
constructor(
|
|
@InjectRepository(Notification)
|
|
private repo: Repository<Notification>,
|
|
private eventEmitter: EventEmitter2,
|
|
) {}
|
|
|
|
async create(dto: CreateNotificationDto): Promise<Notification[]> {
|
|
const recipientIds = [...new Set(dto.recipientIds)];
|
|
if (recipientIds.length === 0) throw new BadRequestException('通知接收人不能为空');
|
|
|
|
const notifications = recipientIds.map((recipientId) => ({
|
|
recipientId,
|
|
type: dto.type,
|
|
title: dto.title,
|
|
content: dto.content ?? '',
|
|
link: dto.link,
|
|
}));
|
|
const saved = await this.repo.save(notifications);
|
|
|
|
// Push SSE + emit event
|
|
for (const n of saved) {
|
|
this.subjects.get(n.recipientId)?.next(n);
|
|
this.eventEmitter.emit('notification.created', n);
|
|
}
|
|
|
|
return saved;
|
|
}
|
|
|
|
async findByUser(userId: number, after?: number, limit: number = 20): Promise<Notification[]> {
|
|
const safeLimit = Math.min(Math.max(limit, 1), 100);
|
|
const qb = this.repo
|
|
.createQueryBuilder('n')
|
|
.where('n.recipientId = :userId', { userId })
|
|
.orderBy('n.createdAt', 'DESC')
|
|
.take(safeLimit);
|
|
|
|
if (after !== undefined) {
|
|
qb.andWhere('n.id < :after', { after });
|
|
}
|
|
|
|
return qb.getMany();
|
|
}
|
|
|
|
async getUnreadCount(userId: number): Promise<number> {
|
|
return this.repo.count({
|
|
where: { recipientId: userId, isRead: false },
|
|
});
|
|
}
|
|
|
|
async markRead(id: number, userId: number): Promise<void> {
|
|
const notification = await this.repo.findOne({ where: { id, recipientId: userId } });
|
|
if (!notification) throw new NotFoundException('通知不存在');
|
|
if (notification.isRead) return;
|
|
await this.repo.update({ id, recipientId: userId }, { isRead: true, readAt: new Date() });
|
|
}
|
|
|
|
async markAllRead(userId: number): Promise<void> {
|
|
await this.repo.update(
|
|
{ recipientId: userId, isRead: false },
|
|
{ isRead: true, readAt: new Date() },
|
|
);
|
|
}
|
|
|
|
subscribe(userId: number): Observable<Notification> {
|
|
if (!this.subjects.has(userId)) {
|
|
this.subjects.set(userId, new Subject<Notification>());
|
|
this.subscriberCounts.set(userId, 0);
|
|
}
|
|
this.subscriberCounts.set(userId, (this.subscriberCounts.get(userId) ?? 0) + 1);
|
|
return this.subjects.get(userId)!.asObservable();
|
|
}
|
|
|
|
unsubscribe(userId: number): void {
|
|
const count = (this.subscriberCounts.get(userId) ?? 0) - 1;
|
|
if (count > 0) {
|
|
this.subscriberCounts.set(userId, count);
|
|
return;
|
|
}
|
|
// Last subscriber gone → complete + clean up
|
|
this.subscriberCounts.delete(userId);
|
|
const subj = this.subjects.get(userId);
|
|
if (subj) {
|
|
subj.complete();
|
|
this.subjects.delete(userId);
|
|
}
|
|
}
|
|
}
|