userFollowingChannelsCacheの場所をCacheServiceからChannelFollowingServiceに移動

packages/backend/src/core/CacheService.ts

@@ -5,7 +5,7 @@
 
 import { Inject, Injectable } from '@nestjs/common';
 import * as Redis from 'ioredis';
-import type { BlockingsRepository, ChannelFollowingsRepository, FollowingsRepository, MutingsRepository, RenoteMutingsRepository, MiUserProfile, UserProfilesRepository, UsersRepository, MiFollowing } from '@/models/_.js';
+import type { BlockingsRepository, FollowingsRepository, MutingsRepository, RenoteMutingsRepository, MiUserProfile, UserProfilesRepository, UsersRepository, MiFollowing } from '@/models/_.js';
 import { MemoryKVCache, RedisKVCache } from '@/misc/cache.js';
 import type { MiLocalUser, MiUser } from '@/models/User.js';
 import { DI } from '@/di-symbols.js';
@@ -26,7 +26,6 @@ export class CacheService implements OnApplicationShutdown {
 	public userBlockedCache: RedisKVCache<Set<string>>; // NOTE: 「被」Blockキャッシュ
 	public renoteMutingsCache: RedisKVCache<Set<string>>;
 	public userFollowingsCache: RedisKVCache<Record<string, Pick<MiFollowing, 'withReplies'> | undefined>>;
-	public userFollowingChannelsCache: RedisKVCache<Set<string>>;
 
 	constructor(
 		@Inject(DI.redis)
@@ -53,9 +52,6 @@ export class CacheService implements OnApplicationShutdown {
 		@Inject(DI.followingsRepository)
 		private followingsRepository: FollowingsRepository,
 
-		@Inject(DI.channelFollowingsRepository)
-		private channelFollowingsRepository: ChannelFollowingsRepository,
-
 		private userEntityService: UserEntityService,
 	) {
 		//this.onMessage = this.onMessage.bind(this);
@@ -150,13 +146,7 @@ export class CacheService implements OnApplicationShutdown {
 			fromRedisConverter: (value) => JSON.parse(value),
 		});
 
-		this.userFollowingChannelsCache = new RedisKVCache<Set<string>>(this.redisClient, 'userFollowingChannels', {
-			lifetime: 1000 * 60 * 30, // 30m
-			memoryCacheLifetime: 1000 * 60, // 1m
-			fetcher: (key) => this.channelFollowingsRepository.find({ where: { followerId: key }, select: ['followeeId'] }).then(xs => new Set(xs.map(x => x.followeeId))),
-			toRedisConverter: (value) => JSON.stringify(Array.from(value)),
-			fromRedisConverter: (value) => new Set(JSON.parse(value)),
-		});
+		// NOTE: チャンネルのフォロー状況キャッシュはChannelFollowingServiceで行っている
 
 		this.redisForSub.on('message', this.onMessage);
 	}
@@ -197,14 +187,6 @@ export class CacheService implements OnApplicationShutdown {
 					this.userFollowingsCache.delete(body.followerId);
 					break;
 				}
-				case 'followChannel': {
-					this.userFollowingChannelsCache.refresh(body.userId);
-					break;
-				}
-				case 'unfollowChannel': {
-					this.userFollowingChannelsCache.delete(body.userId);
-					break;
-				}
 				default:
 					break;
 			}
@@ -229,7 +211,6 @@ export class CacheService implements OnApplicationShutdown {
 		this.userBlockedCache.dispose();
 		this.renoteMutingsCache.dispose();
 		this.userFollowingsCache.dispose();
-		this.userFollowingChannelsCache.dispose();
 	}
 
 	@bindThis

packages/backend/src/core/ChannelFollowingService.ts

@@ -1,22 +1,40 @@
 import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
+import Redis from 'ioredis';
 import { DI } from '@/di-symbols.js';
 import type { ChannelFollowingsRepository } from '@/models/_.js';
 import { MiChannel } from '@/models/_.js';
 import { IdService } from '@/core/IdService.js';
-import { CacheService } from '@/core/CacheService.js';
-import { GlobalEventService } from '@/core/GlobalEventService.js';
+import { GlobalEvents, GlobalEventService } from '@/core/GlobalEventService.js';
 import { bindThis } from '@/decorators.js';
 import type { MiLocalUser } from '@/models/User.js';
+import { RedisKVCache } from '@/misc/cache.js';
 
 @Injectable()
 export class ChannelFollowingService implements OnModuleInit {
+	public userFollowingChannelsCache: RedisKVCache<Set<string>>;
+
 	constructor(
+		@Inject(DI.redis)
+		private redisClient: Redis.Redis,
+		@Inject(DI.redisForSub)
+		private redisForSub: Redis.Redis,
 		@Inject(DI.channelFollowingsRepository)
 		private channelFollowingsRepository: ChannelFollowingsRepository,
 		private idService: IdService,
-		private cacheService: CacheService,
 		private globalEventService: GlobalEventService,
 	) {
+		this.userFollowingChannelsCache = new RedisKVCache<Set<string>>(this.redisClient, 'userFollowingChannels', {
+			lifetime: 1000 * 60 * 30, // 30m
+			memoryCacheLifetime: 1000 * 60, // 1m
+			fetcher: (key) => this.channelFollowingsRepository.find({
+				where: { followerId: key },
+				select: ['followeeId'],
+			}).then(xs => new Set(xs.map(x => x.followeeId))),
+			toRedisConverter: (value) => JSON.stringify(Array.from(value)),
+			fromRedisConverter: (value) => new Set(JSON.parse(value)),
+		});
+
+		this.redisForSub.on('message', this.onMessage);
 	}
 
 	onModuleInit() {
@@ -33,8 +51,6 @@ export class ChannelFollowingService implements OnModuleInit {
 			followeeId: targetChannel.id,
 		});
 
-		this.cacheService.userFollowingsCache.refresh(requestUser.id);
-
 		this.globalEventService.publishInternalEvent('followChannel', {
 			userId: requestUser.id,
 			channelId: targetChannel.id,
@@ -51,11 +67,38 @@ export class ChannelFollowingService implements OnModuleInit {
 			followeeId: targetChannel.id,
 		});
 
-		this.cacheService.userFollowingsCache.refresh(requestUser.id);
-
 		this.globalEventService.publishInternalEvent('unfollowChannel', {
 			userId: requestUser.id,
 			channelId: targetChannel.id,
 		});
 	}
+
+	@bindThis
+	private async onMessage(_: string, data: string): Promise<void> {
+		const obj = JSON.parse(data);
+
+		if (obj.channel === 'internal') {
+			const { type, body } = obj.message as GlobalEvents['internal']['payload'];
+			switch (type) {
+				case 'followChannel': {
+					this.userFollowingChannelsCache.refresh(body.userId);
+					break;
+				}
+				case 'unfollowChannel': {
+					this.userFollowingChannelsCache.delete(body.userId);
+					break;
+				}
+			}
+		}
+	}
+
+	@bindThis
+	public dispose(): void {
+		this.userFollowingChannelsCache.dispose();
+	}
+
+	@bindThis
+	public onApplicationShutdown(signal?: string | undefined): void {
+		this.dispose();
+	}
 }

packages/backend/src/server/api/StreamingApiServerService.ts

@@ -15,6 +15,7 @@ import { bindThis } from '@/decorators.js';
 import { CacheService } from '@/core/CacheService.js';
 import { MiLocalUser } from '@/models/User.js';
 import { UserService } from '@/core/UserService.js';
+import { ChannelFollowingService } from '@/core/ChannelFollowingService.js';
 import { AuthenticateService, AuthenticationError } from './AuthenticateService.js';
 import MainStreamConnection from './stream/Connection.js';
 import { ChannelsService } from './stream/ChannelsService.js';
@@ -39,6 +40,7 @@ export class StreamingApiServerService {
 		private channelsService: ChannelsService,
 		private notificationService: NotificationService,
 		private usersService: UserService,
+		private channelFollowingService: ChannelFollowingService,
 	) {
 	}
 
@@ -93,6 +95,7 @@ export class StreamingApiServerService {
 				this.noteReadService,
 				this.notificationService,
 				this.cacheService,
+				this.channelFollowingService,
 				user, app,
 			);
 

packages/backend/src/server/api/stream/Connection.ts

@@ -13,6 +13,7 @@ import { bindThis } from '@/decorators.js';
 import { CacheService } from '@/core/CacheService.js';
 import { MiFollowing, MiUserProfile } from '@/models/_.js';
 import type { StreamEventEmitter, GlobalEvents } from '@/core/GlobalEventService.js';
+import { ChannelFollowingService } from '@/core/ChannelFollowingService.js';
 import type { ChannelsService } from './ChannelsService.js';
 import type { EventEmitter } from 'events';
 import type Channel from './channel.js';
@@ -42,6 +43,7 @@ export default class Connection {
 		private noteReadService: NoteReadService,
 		private notificationService: NotificationService,
 		private cacheService: CacheService,
+		private channelFollowingService: ChannelFollowingService,
 
 		user: MiUser | null | undefined,
 		token: MiAccessToken | null | undefined,
@@ -56,7 +58,7 @@ export default class Connection {
 		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.channelFollowingService.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),