From f44504097c360fc84179161abee47b79a936b455 Mon Sep 17 00:00:00 2001 From: syuilo Date: Wed, 5 Apr 2023 10:21:10 +0900 Subject: enhance(backend): improve cache --- packages/backend/src/server/api/stream/channel.ts | 12 +- .../src/server/api/stream/channels/antenna.ts | 6 +- .../src/server/api/stream/channels/channel.ts | 6 +- .../server/api/stream/channels/global-timeline.ts | 6 +- .../src/server/api/stream/channels/hashtag.ts | 6 +- .../server/api/stream/channels/home-timeline.ts | 11 +- .../server/api/stream/channels/hybrid-timeline.ts | 8 +- .../server/api/stream/channels/local-timeline.ts | 6 +- .../backend/src/server/api/stream/channels/main.ts | 4 +- .../src/server/api/stream/channels/user-list.ts | 6 +- packages/backend/src/server/api/stream/index.ts | 187 +++++---------------- packages/backend/src/server/api/stream/types.ts | 21 +-- 12 files changed, 82 insertions(+), 197 deletions(-) (limited to 'packages/backend/src/server/api/stream') diff --git a/packages/backend/src/server/api/stream/channel.ts b/packages/backend/src/server/api/stream/channel.ts index 32935325aa..e67aec9ecd 100644 --- a/packages/backend/src/server/api/stream/channel.ts +++ b/packages/backend/src/server/api/stream/channel.ts @@ -23,16 +23,16 @@ export default abstract class Channel { return this.connection.following; } - protected get muting() { - return this.connection.muting; + protected get userIdsWhoMeMuting() { + return this.connection.userIdsWhoMeMuting; } - protected get renoteMuting() { - return this.connection.renoteMuting; + protected get userIdsWhoMeMutingRenotes() { + return this.connection.userIdsWhoMeMutingRenotes; } - protected get blocking() { - return this.connection.blocking; + protected get userIdsWhoBlockingMe() { + return this.connection.userIdsWhoBlockingMe; } protected get followingChannels() { diff --git a/packages/backend/src/server/api/stream/channels/antenna.ts b/packages/backend/src/server/api/stream/channels/antenna.ts index e2a42fbfe9..d48dea7258 100644 --- a/packages/backend/src/server/api/stream/channels/antenna.ts +++ b/packages/backend/src/server/api/stream/channels/antenna.ts @@ -35,11 +35,11 @@ class AntennaChannel extends Channel { const note = await this.noteEntityService.pack(data.body.id, this.user, { detail: true }); // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; this.connection.cacheNote(note); diff --git a/packages/backend/src/server/api/stream/channels/channel.ts b/packages/backend/src/server/api/stream/channels/channel.ts index 12caa7f233..9e5b40997b 100644 --- a/packages/backend/src/server/api/stream/channels/channel.ts +++ b/packages/backend/src/server/api/stream/channels/channel.ts @@ -47,11 +47,11 @@ class ChannelChannel extends Channel { } // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; this.connection.cacheNote(note); diff --git a/packages/backend/src/server/api/stream/channels/global-timeline.ts b/packages/backend/src/server/api/stream/channels/global-timeline.ts index d79247cd6e..5454836fe1 100644 --- a/packages/backend/src/server/api/stream/channels/global-timeline.ts +++ b/packages/backend/src/server/api/stream/channels/global-timeline.ts @@ -64,11 +64,11 @@ class GlobalTimelineChannel extends Channel { if (isInstanceMuted(note, new Set(this.userProfile?.mutedInstances ?? []))) return; // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; // 流れてきたNoteがミュートすべきNoteだったら無視する // TODO: 将来的には、単にMutedNoteテーブルにレコードがあるかどうかで判定したい(以下の理由により難しそうではある) diff --git a/packages/backend/src/server/api/stream/channels/hashtag.ts b/packages/backend/src/server/api/stream/channels/hashtag.ts index 98dc858ded..0268fdedde 100644 --- a/packages/backend/src/server/api/stream/channels/hashtag.ts +++ b/packages/backend/src/server/api/stream/channels/hashtag.ts @@ -46,11 +46,11 @@ class HashtagChannel extends Channel { } // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; this.connection.cacheNote(note); diff --git a/packages/backend/src/server/api/stream/channels/home-timeline.ts b/packages/backend/src/server/api/stream/channels/home-timeline.ts index c623fef64a..ee874ad81e 100644 --- a/packages/backend/src/server/api/stream/channels/home-timeline.ts +++ b/packages/backend/src/server/api/stream/channels/home-timeline.ts @@ -24,7 +24,6 @@ class HomeTimelineChannel extends Channel { @bindThis public async init(params: any) { - // Subscribe events this.subscriber.on('notesStream', this.onNote); } @@ -38,7 +37,7 @@ class HomeTimelineChannel extends Channel { } // Ignore notes from instances the user has muted - if (isInstanceMuted(note, new Set(this.userProfile?.mutedInstances ?? []))) return; + if (isInstanceMuted(note, new Set(this.userProfile!.mutedInstances ?? []))) return; if (['followers', 'specified'].includes(note.visibility)) { note = await this.noteEntityService.pack(note.id, this.user!, { @@ -71,18 +70,18 @@ class HomeTimelineChannel extends Channel { } // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; // 流れてきたNoteがミュートすべきNoteだったら無視する // TODO: 将来的には、単にMutedNoteテーブルにレコードがあるかどうかで判定したい(以下の理由により難しそうではある) // 現状では、ワードミュートにおけるMutedNoteレコードの追加処理はストリーミングに流す処理と並列で行われるため、 // レコードが追加されるNoteでも追加されるより先にここのストリーミングの処理に到達することが起こる。 // そのためレコードが存在するかのチェックでは不十分なので、改めてcheckWordMuteを呼んでいる - if (this.userProfile && await checkWordMute(note, this.user, this.userProfile.mutedWords)) return; + if (await checkWordMute(note, this.user, this.userProfile!.mutedWords)) return; this.connection.cacheNote(note); diff --git a/packages/backend/src/server/api/stream/channels/hybrid-timeline.ts b/packages/backend/src/server/api/stream/channels/hybrid-timeline.ts index f54767bc9d..4f7b4e78b6 100644 --- a/packages/backend/src/server/api/stream/channels/hybrid-timeline.ts +++ b/packages/backend/src/server/api/stream/channels/hybrid-timeline.ts @@ -72,7 +72,7 @@ class HybridTimelineChannel extends Channel { } // Ignore notes from instances the user has muted - if (isInstanceMuted(note, new Set(this.userProfile?.mutedInstances ?? []))) return; + if (isInstanceMuted(note, new Set(this.userProfile!.mutedInstances ?? []))) return; // 関係ない返信は除外 if (note.reply && !this.user!.showTimelineReplies) { @@ -82,11 +82,11 @@ class HybridTimelineChannel extends Channel { } // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; // 流れてきたNoteがミュートすべきNoteだったら無視する // TODO: 将来的には、単にMutedNoteテーブルにレコードがあるかどうかで判定したい(以下の理由により難しそうではある) diff --git a/packages/backend/src/server/api/stream/channels/local-timeline.ts b/packages/backend/src/server/api/stream/channels/local-timeline.ts index eb0642900d..836c5aae6c 100644 --- a/packages/backend/src/server/api/stream/channels/local-timeline.ts +++ b/packages/backend/src/server/api/stream/channels/local-timeline.ts @@ -61,11 +61,11 @@ class LocalTimelineChannel extends Channel { } // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; // 流れてきたNoteがミュートすべきNoteだったら無視する // TODO: 将来的には、単にMutedNoteテーブルにレコードがあるかどうかで判定したい(以下の理由により難しそうではある) diff --git a/packages/backend/src/server/api/stream/channels/main.ts b/packages/backend/src/server/api/stream/channels/main.ts index 4dd16b530a..139320ce35 100644 --- a/packages/backend/src/server/api/stream/channels/main.ts +++ b/packages/backend/src/server/api/stream/channels/main.ts @@ -26,7 +26,7 @@ class MainChannel extends Channel { case 'notification': { // Ignore notifications from instances the user has muted if (isUserFromMutedInstance(data.body, new Set(this.userProfile?.mutedInstances ?? []))) return; - if (data.body.userId && this.muting.has(data.body.userId)) return; + if (data.body.userId && this.userIdsWhoMeMuting.has(data.body.userId)) return; if (data.body.note && data.body.note.isHidden) { const note = await this.noteEntityService.pack(data.body.note.id, this.user, { @@ -40,7 +40,7 @@ class MainChannel extends Channel { case 'mention': { if (isInstanceMuted(data.body, new Set(this.userProfile?.mutedInstances ?? []))) return; - if (this.muting.has(data.body.userId)) return; + if (this.userIdsWhoMeMuting.has(data.body.userId)) return; if (data.body.isHidden) { const note = await this.noteEntityService.pack(data.body.id, this.user, { detail: true, diff --git a/packages/backend/src/server/api/stream/channels/user-list.ts b/packages/backend/src/server/api/stream/channels/user-list.ts index 8a42e99a54..8802fc5ab8 100644 --- a/packages/backend/src/server/api/stream/channels/user-list.ts +++ b/packages/backend/src/server/api/stream/channels/user-list.ts @@ -89,11 +89,11 @@ class UserListChannel extends Channel { } // 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.muting)) return; + if (isUserRelated(note, this.userIdsWhoMeMuting)) return; // 流れてきたNoteがブロックされているユーザーが関わるものだったら無視する - if (isUserRelated(note, this.blocking)) return; + if (isUserRelated(note, this.userIdsWhoBlockingMe)) return; - if (note.renote && !note.text && isUserRelated(note, this.renoteMuting)) return; + if (note.renote && !note.text && isUserRelated(note, this.userIdsWhoMeMutingRenotes)) return; this.send('note', note); } diff --git a/packages/backend/src/server/api/stream/index.ts b/packages/backend/src/server/api/stream/index.ts index 2f473cd012..a6f9145952 100644 --- a/packages/backend/src/server/api/stream/index.ts +++ b/packages/backend/src/server/api/stream/index.ts @@ -1,13 +1,11 @@ import type { User } from '@/models/entities/User.js'; -import type { Channel as ChannelModel } from '@/models/entities/Channel.js'; -import type { FollowingsRepository, MutingsRepository, RenoteMutingsRepository, UserProfilesRepository, ChannelFollowingsRepository, BlockingsRepository } from '@/models/index.js'; import type { AccessToken } from '@/models/entities/AccessToken.js'; -import type { UserProfile } from '@/models/entities/UserProfile.js'; import type { Packed } from '@/misc/json-schema.js'; -import type { GlobalEventService } from '@/core/GlobalEventService.js'; import type { NoteReadService } from '@/core/NoteReadService.js'; import type { NotificationService } from '@/core/NotificationService.js'; import { bindThis } from '@/decorators.js'; +import { CacheService } from '@/core/CacheService.js'; +import { UserProfile } from '@/models/index.js'; import type { ChannelsService } from './ChannelsService.js'; import type * as websocket from 'websocket'; import type { EventEmitter } from 'events'; @@ -19,106 +17,71 @@ import type { StreamEventEmitter, StreamMessages } from './types.js'; */ export default class Connection { public user?: User; - public userProfile?: UserProfile | null; - public following: Set = new Set(); - public muting: Set = new Set(); - public renoteMuting: Set = new Set(); - public blocking: Set = new Set(); // "被"blocking - public followingChannels: Set = new Set(); public token?: AccessToken; private wsConnection: websocket.connection; public subscriber: StreamEventEmitter; private channels: Channel[] = []; private subscribingNotes: any = {}; private cachedNotes: Packed<'Note'>[] = []; + public userProfile: UserProfile | null = null; + public following: Set = new Set(); + public followingChannels: Set = new Set(); + public userIdsWhoMeMuting: Set = new Set(); + public userIdsWhoBlockingMe: Set = new Set(); + public userIdsWhoMeMutingRenotes: Set = new Set(); + private fetchIntervalId: NodeJS.Timer | null = null; constructor( - private followingsRepository: FollowingsRepository, - private mutingsRepository: MutingsRepository, - private renoteMutingsRepository: RenoteMutingsRepository, - private blockingsRepository: BlockingsRepository, - private channelFollowingsRepository: ChannelFollowingsRepository, - private userProfilesRepository: UserProfilesRepository, private channelsService: ChannelsService, - private globalEventService: GlobalEventService, private noteReadService: NoteReadService, private notificationService: NotificationService, + private cacheService: CacheService, - wsConnection: websocket.connection, subscriber: EventEmitter, user: User | null | undefined, token: AccessToken | null | undefined, ) { - this.wsConnection = wsConnection; this.subscriber = subscriber; if (user) this.user = user; if (token) this.token = token; + } - //this.onWsConnectionMessage = this.onWsConnectionMessage.bind(this); - //this.onUserEvent = this.onUserEvent.bind(this); - //this.onNoteStreamMessage = this.onNoteStreamMessage.bind(this); - //this.onBroadcastMessage = this.onBroadcastMessage.bind(this); - - this.wsConnection.on('message', this.onWsConnectionMessage); - - this.subscriber.on('broadcast', data => { - this.onBroadcastMessage(data); - }); + @bindThis + public async fetch() { + if (this.user == null) return; + const [userProfile, following, followingChannels, userIdsWhoMeMuting, userIdsWhoBlockingMe, userIdsWhoMeMutingRenotes] = await Promise.all([ + this.cacheService.userProfileCache.fetch(this.user.id), + this.cacheService.userFollowingsCache.fetch(this.user.id), + this.cacheService.userFollowingChannelsCache.fetch(this.user.id), + this.cacheService.userMutingsCache.fetch(this.user.id), + this.cacheService.userBlockedCache.fetch(this.user.id), + this.cacheService.renoteMutingsCache.fetch(this.user.id), + ]); + this.userProfile = userProfile; + this.following = following; + this.followingChannels = followingChannels; + this.userIdsWhoMeMuting = userIdsWhoMeMuting; + this.userIdsWhoBlockingMe = userIdsWhoBlockingMe; + this.userIdsWhoMeMutingRenotes = userIdsWhoMeMutingRenotes; + } - if (this.user) { - this.updateFollowing(); - this.updateMuting(); - this.updateRenoteMuting(); - this.updateBlocking(); - this.updateFollowingChannels(); - this.updateUserProfile(); + @bindThis + public async init() { + if (this.user != null) { + await this.fetch(); - this.subscriber.on(`user:${this.user.id}`, this.onUserEvent); + this.fetchIntervalId = setInterval(this.fetch, 1000 * 10); } } @bindThis - private onUserEvent(data: StreamMessages['user']['payload']) { // { type, body }と展開するとそれぞれ型が分離してしまう - switch (data.type) { - case 'follow': - this.following.add(data.body.id); - break; - - case 'unfollow': - this.following.delete(data.body.id); - break; - - case 'mute': - this.muting.add(data.body.id); - break; - - case 'unmute': - this.muting.delete(data.body.id); - break; - - // TODO: renote mute events - // TODO: block events - - case 'followChannel': - this.followingChannels.add(data.body.id); - break; - - case 'unfollowChannel': - this.followingChannels.delete(data.body.id); - break; - - case 'updateUserProfile': - this.userProfile = data.body; - break; - - case 'terminate': - this.wsConnection.close(); - this.dispose(); - break; - - default: - break; - } + public async init2(wsConnection: websocket.connection) { + this.wsConnection = wsConnection; + this.wsConnection.on('message', this.onWsConnectionMessage); + + this.subscriber.on('broadcast', data => { + this.onBroadcastMessage(data); + }); } /** @@ -318,78 +281,12 @@ export default class Connection { } } - @bindThis - private async updateFollowing() { - const followings = await this.followingsRepository.find({ - where: { - followerId: this.user!.id, - }, - select: ['followeeId'], - }); - - this.following = new Set(followings.map(x => x.followeeId)); - } - - @bindThis - private async updateMuting() { - const mutings = await this.mutingsRepository.find({ - where: { - muterId: this.user!.id, - }, - select: ['muteeId'], - }); - - this.muting = new Set(mutings.map(x => x.muteeId)); - } - - @bindThis - private async updateRenoteMuting() { - const renoteMutings = await this.renoteMutingsRepository.find({ - where: { - muterId: this.user!.id, - }, - select: ['muteeId'], - }); - - this.renoteMuting = new Set(renoteMutings.map(x => x.muteeId)); - } - - @bindThis - private async updateBlocking() { // ここでいうBlockingは被Blockingの意 - const blockings = await this.blockingsRepository.find({ - where: { - blockeeId: this.user!.id, - }, - select: ['blockerId'], - }); - - this.blocking = new Set(blockings.map(x => x.blockerId)); - } - - @bindThis - private async updateFollowingChannels() { - const followings = await this.channelFollowingsRepository.find({ - where: { - followerId: this.user!.id, - }, - select: ['followeeId'], - }); - - this.followingChannels = new Set(followings.map(x => x.followeeId)); - } - - @bindThis - private async updateUserProfile() { - this.userProfile = await this.userProfilesRepository.findOneBy({ - userId: this.user!.id, - }); - } - /** * ストリームが切れたとき */ @bindThis public dispose() { + if (this.fetchIntervalId) clearInterval(this.fetchIntervalId); for (const c of this.channels.filter(c => c.dispose)) { if (c.dispose) c.dispose(); } diff --git a/packages/backend/src/server/api/stream/types.ts b/packages/backend/src/server/api/stream/types.ts index f4eedc3964..ed73897e73 100644 --- a/packages/backend/src/server/api/stream/types.ts +++ b/packages/backend/src/server/api/stream/types.ts @@ -38,6 +38,11 @@ export interface InternalStreamTypes { antennaDeleted: Antenna; antennaUpdated: Antenna; metaUpdated: Meta; + followChannel: { userId: User['id']; channelId: Channel['id']; }; + unfollowChannel: { userId: User['id']; channelId: Channel['id']; }; + updateUserProfile: UserProfile; + mute: { muterId: User['id']; muteeId: User['id']; }; + unmute: { muterId: User['id']; muteeId: User['id']; }; } export interface BroadcastTypes { @@ -56,18 +61,6 @@ export interface BroadcastTypes { }; } -export interface UserStreamTypes { - terminate: Record; - followChannel: Channel; - unfollowChannel: Channel; - updateUserProfile: UserProfile; - mute: User; - unmute: User; - follow: Packed<'UserDetailedNotMe'>; - unfollow: Packed<'User'>; - userAdded: Packed<'User'>; -} - export interface MainStreamTypes { notification: Packed<'Notification'>; mention: Packed<'Note'>; @@ -200,10 +193,6 @@ export type StreamMessages = { name: 'broadcast'; payload: EventUnionFromDictionary>; }; - user: { - name: `user:${User['id']}`; - payload: EventUnionFromDictionary>; - }; main: { name: `mainStream:${User['id']}`; payload: EventUnionFromDictionary>; -- cgit v1.2.3-freya