oscar-surf/misskey
oscar-surf/misskey
0
0
Fork 0
Code Issues Activity

perf(backend): 通知をRedisに保存するように

Browse source 
Resolve #10168
This commit is contained in:
syuilo 2023-04-04 14:06:57 +09:00
parent 38d0b62167
commit 30d6992684
29 changed files with 185 additions and 613 deletions
Download patch file Download diff file Expand all files Collapse all files

117
packages/backend/src/core/NotificationService.ts
View file

@ -1,8 +1,9 @@
import { setTimeout } from 'node:timers/promises';
import Redis from 'ioredis';
import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common';
import { In } from 'typeorm';
import { DI } from '@/di-symbols.js';
import type { MutingsRepository, NotificationsRepository, UserProfilesRepository, UsersRepository } from '@/models/index.js';
import type { MutingsRepository, UserProfilesRepository, UsersRepository } from '@/models/index.js';
import type { User } from '@/models/entities/User.js';
import type { Notification } from '@/models/entities/Notification.js';
import { UserEntityService } from '@/core/entities/UserEntityService.js';
@ -17,15 +18,15 @@ export class NotificationService implements OnApplicationShutdown {
#shutdownController = new AbortController();
constructor(
@Inject(DI.redis)
private redisClient: Redis.Redis,
@Inject(DI.usersRepository)
private usersRepository: UsersRepository,
@Inject(DI.userProfilesRepository)
private userProfilesRepository: UserProfilesRepository,
@Inject(DI.notificationsRepository)
private notificationsRepository: NotificationsRepository,
@Inject(DI.mutingsRepository)
private mutingsRepository: MutingsRepository,
@ -38,50 +39,31 @@ export class NotificationService implements OnApplicationShutdown {
}
@bindThis
public async readNotification(
public async readAllNotification(
userId: User['id'],
notificationIds: Notification['id'][],
) {
if (notificationIds.length === 0) return;
const latestReadNotificationId = await this.redisClient.get(`latestReadNotification:${userId}`);
const latestNotificationIdsRes = await this.redisClient.xrevrange(
`notificationTimeline:${userId}`,
'+',
'-',
'COUNT', 1);
console.log('latestNotificationIdsRes', latestNotificationIdsRes);
const latestNotificationId = latestNotificationIdsRes[0]?.[0];
// Update documents
const result = await this.notificationsRepository.update({
notifieeId: userId,
id: In(notificationIds),
isRead: false,
}, {
isRead: true,
});
if (latestNotificationId == null) return;
if (result.affected === 0) return;
this.redisClient.set(`latestReadNotification:${userId}`, latestNotificationId);
if (!await this.userEntityService.getHasUnreadNotification(userId)) return this.postReadAllNotifications(userId);
else return this.postReadNotifications(userId, notificationIds);
}
@bindThis
public async readNotificationByQuery(
userId: User['id'],
query: Record<string, any>,
) {
const notificationIds = await this.notificationsRepository.findBy({
...query,
notifieeId: userId,
isRead: false,
}).then(notifications => notifications.map(notification => notification.id));
return this.readNotification(userId, notificationIds);
if (latestReadNotificationId == null || (latestReadNotificationId < latestNotificationId)) {
return this.postReadAllNotifications(userId);
}
}
@bindThis
private postReadAllNotifications(userId: User['id']) {
this.globalEventService.publishMainStream(userId, 'readAllNotifications');
return this.pushNotificationService.pushNotification(userId, 'readAllNotifications', undefined);
}
@bindThis
private postReadNotifications(userId: User['id'], notificationIds: Notification['id'][]) {
return this.pushNotificationService.pushNotification(userId, 'readNotifications', { notificationIds });
}
@bindThis
@ -90,47 +72,48 @@ export class NotificationService implements OnApplicationShutdown {
type: Notification['type'],
data: Partial<Notification>,
): Promise<Notification | null> {
if (data.notifierId && (notifieeId === data.notifierId)) {
return null;
// TODO: Cache
const profile = await this.userProfilesRepository.findOneBy({ userId: notifieeId });
const isMuted = profile?.mutingNotificationTypes.includes(type);
if (isMuted) return null;
if (data.notifierId) {
if (notifieeId === data.notifierId) {
return null;
}
// TODO: cache
const mutings = await this.mutingsRepository.findOneBy({
muterId: notifieeId,
muteeId: data.notifierId,
});
if (mutings) {
return null;
}
}
const profile = await this.userProfilesRepository.findOneBy({ userId: notifieeId });
// TODO: Cache
const isMuted = profile?.mutingNotificationTypes.includes(type);
// Create notification
const notification = await this.notificationsRepository.insert({
const notification = {
id: this.idService.genId(),
createdAt: new Date(),
notifieeId: notifieeId,
type: type,
// 相手がこの通知をミュートしているようなら、既読を予めつけておく
isRead: isMuted,
...data,
} as Partial<Notification>)
.then(x => this.notificationsRepository.findOneByOrFail(x.identifiers[0]));
} as Notification;
const packed = await this.notificationEntityService.pack(notification, {});
this.redisClient.xadd(
`notificationTimeline:${notifieeId}`,
'MAXLEN', '~', '300',
`${this.idService.parse(notification.id).date.getTime()}-*`,
'data', JSON.stringify(notification));
const packed = await this.notificationEntityService.pack(notification, notifieeId, {});
// Publish notification event
this.globalEventService.publishMainStream(notifieeId, 'notification', packed);
// 2秒経っても(今回作成した)通知が既読にならなかったら「未読の通知がありますよ」イベントを発行する
setTimeout(2000, 'unread note', { signal: this.#shutdownController.signal }).then(async () => {
const fresh = await this.notificationsRepository.findOneBy({ id: notification.id });
if (fresh == null) return; // 既に削除されているかもしれない
if (fresh.isRead) return;
//#region ただしミュートしているユーザーからの通知なら無視
// TODO: Cache
const mutings = await this.mutingsRepository.findBy({
muterId: notifieeId,
});
if (data.notifierId && mutings.map(m => m.muteeId).includes(data.notifierId)) {
return;
}
//#endregion
setTimeout(2000, 'unread notification', { signal: this.#shutdownController.signal }).then(async () => {
const latestReadNotificationId = await this.redisClient.get(`latestReadNotification:${notifieeId}`);
if (latestReadNotificationId && (latestReadNotificationId >= notification.id)) return;
this.globalEventService.publishMainStream(notifieeId, 'unreadNotification', packed);
this.pushNotificationService.pushNotification(notifieeId, 'notification', packed);