enhance(backend): improve cache

12 files changed, 81 insertions, 196 deletions

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<string>(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<string>(this.userProfile?.mutedInstances ?? []))) return;
+		if (isInstanceMuted(note, new Set<string>(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<string>(this.userProfile?.mutedInstances ?? []))) return;
+		if (isInstanceMuted(note, new Set<string>(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<string>(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<string>(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<User['id']> = new Set();
-	public muting: Set<User['id']> = new Set();
-	public renoteMuting: Set<User['id']> = new Set();
-	public blocking: Set<User['id']> = new Set(); // "被"blocking
-	public followingChannels: Set<ChannelModel['id']> = 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<string> = new Set();
+	public followingChannels: Set<string> = new Set();
+	public userIdsWhoMeMuting: Set<string> = new Set();
+	public userIdsWhoBlockingMe: Set<string> = new Set();
+	public userIdsWhoMeMutingRenotes: Set<string> = 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;
+	public async init2(wsConnection: websocket.connection) {
+		this.wsConnection = wsConnection;
+		this.wsConnection.on('message', this.onWsConnectionMessage);
 
-			default:
-				break;
-		}
+		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<string>(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<string>(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<string>(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<string>(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<string>(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<string, unknown>;
-	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<SerializedAll<BroadcastTypes>>;
 	};
-	user: {
-		name: `user:${User['id']}`;
-		payload: EventUnionFromDictionary<SerializedAll<UserStreamTypes>>;
-	};
 	main: {
 		name: `mainStream:${User['id']}`;
 		payload: EventUnionFromDictionary<SerializedAll<MainStreamTypes>>;