diff --git a/packages/backend/src/core/CacheService.ts b/packages/backend/src/core/CacheService.ts index 3525846785..e1413342b1 100644 --- a/packages/backend/src/core/CacheService.ts +++ b/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>; // NOTE: 「被」Blockキャッシュ public renoteMutingsCache: RedisKVCache>; public userFollowingsCache: RedisKVCache | undefined>>; - public userFollowingChannelsCache: RedisKVCache>; 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>(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 diff --git a/packages/backend/src/core/ChannelFollowingService.ts b/packages/backend/src/core/ChannelFollowingService.ts index 4693177ae5..75843b9773 100644 --- a/packages/backend/src/core/ChannelFollowingService.ts +++ b/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>; + 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>(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 { + 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(); + } } diff --git a/packages/backend/src/server/api/StreamingApiServerService.ts b/packages/backend/src/server/api/StreamingApiServerService.ts index badcec1b33..dc3a00617c 100644 --- a/packages/backend/src/server/api/StreamingApiServerService.ts +++ b/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, ); diff --git a/packages/backend/src/server/api/stream/Connection.ts b/packages/backend/src/server/api/stream/Connection.ts index f981e63871..2d8fec30b1 100644 --- a/packages/backend/src/server/api/stream/Connection.ts +++ b/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),