diff options
| author | syuilo <Syuilotan@yahoo.co.jp> | 2023-04-05 10:21:10 +0900 |
|---|---|---|
| committer | syuilo <Syuilotan@yahoo.co.jp> | 2023-04-05 10:21:10 +0900 |
| commit | f44504097c360fc84179161abee47b79a936b455 (patch) | |
| tree | aec17f0837ccad89754fae24c044ba0b8d7e8def /packages/backend/src/core | |
| parent | enhance(backend): チャンネルの既読管理を削除 (diff) | |
| download | sharkey-f44504097c360fc84179161abee47b79a936b455.tar.gz sharkey-f44504097c360fc84179161abee47b79a936b455.tar.bz2 sharkey-f44504097c360fc84179161abee47b79a936b455.zip | |
enhance(backend): improve cache
Diffstat (limited to 'packages/backend/src/core')
| -rw-r--r-- | packages/backend/src/core/CacheService.ts | 85 | ||||
| -rw-r--r-- | packages/backend/src/core/DeleteAccountService.ts | 3 | ||||
| -rw-r--r-- | packages/backend/src/core/GlobalEventService.ts | 6 | ||||
| -rw-r--r-- | packages/backend/src/core/NotificationService.ts | 6 | ||||
| -rw-r--r-- | packages/backend/src/core/UserBlockingService.ts | 130 | ||||
| -rw-r--r-- | packages/backend/src/core/UserFollowingService.ts | 9 | ||||
| -rw-r--r-- | packages/backend/src/core/UserMutingService.ts | 31 |
7 files changed, 128 insertions, 142 deletions
diff --git a/packages/backend/src/core/CacheService.ts b/packages/backend/src/core/CacheService.ts index 887baeb2c2..f0c311b9b0 100644 --- a/packages/backend/src/core/CacheService.ts +++ b/packages/backend/src/core/CacheService.ts @@ -1,6 +1,6 @@ import { Inject, Injectable } from '@nestjs/common'; import Redis from 'ioredis'; -import type { UserProfile, UsersRepository } from '@/models/index.js'; +import type { BlockingsRepository, ChannelFollowingsRepository, FollowingsRepository, MutingsRepository, RenoteMutingsRepository, UserProfile, UserProfilesRepository, UsersRepository } from '@/models/index.js'; import { MemoryKVCache, RedisKVCache } from '@/misc/cache.js'; import type { LocalUser, User } from '@/models/entities/User.js'; import { DI } from '@/di-symbols.js'; @@ -16,7 +16,12 @@ export class CacheService implements OnApplicationShutdown { public localUserByIdCache: MemoryKVCache<LocalUser>; public uriPersonCache: MemoryKVCache<User | null>; public userProfileCache: RedisKVCache<UserProfile>; - public userMutingsCache: RedisKVCache<string[]>; + public userMutingsCache: RedisKVCache<Set<string>>; + public userBlockingCache: RedisKVCache<Set<string>>; + public userBlockedCache: RedisKVCache<Set<string>>; // NOTE: 「被」Blockキャッシュ + public renoteMutingsCache: RedisKVCache<Set<string>>; + public userFollowingsCache: RedisKVCache<Set<string>>; + public userFollowingChannelsCache: RedisKVCache<Set<string>>; constructor( @Inject(DI.redis) @@ -28,6 +33,24 @@ export class CacheService implements OnApplicationShutdown { @Inject(DI.usersRepository) private usersRepository: UsersRepository, + @Inject(DI.userProfilesRepository) + private userProfilesRepository: UserProfilesRepository, + + @Inject(DI.mutingsRepository) + private mutingsRepository: MutingsRepository, + + @Inject(DI.blockingsRepository) + private blockingsRepository: BlockingsRepository, + + @Inject(DI.renoteMutingsRepository) + private renoteMutingsRepository: RenoteMutingsRepository, + + @Inject(DI.followingsRepository) + private followingsRepository: FollowingsRepository, + + @Inject(DI.channelFollowingsRepository) + private channelFollowingsRepository: ChannelFollowingsRepository, + private userEntityService: UserEntityService, ) { //this.onMessage = this.onMessage.bind(this); @@ -36,8 +59,62 @@ export class CacheService implements OnApplicationShutdown { this.localUserByNativeTokenCache = new MemoryKVCache<LocalUser | null>(Infinity); this.localUserByIdCache = new MemoryKVCache<LocalUser>(Infinity); this.uriPersonCache = new MemoryKVCache<User | null>(Infinity); - this.userProfileCache = new RedisKVCache<UserProfile>(this.redisClient, 'userProfile', 1000 * 60 * 60 * 24, 1000 * 60); - this.userMutingsCache = new RedisKVCache<string[]>(this.redisClient, 'userMutings', 1000 * 60 * 60 * 24, 1000 * 60); + + this.userProfileCache = new RedisKVCache<UserProfile>(this.redisClient, 'userProfile', { + lifetime: 1000 * 60 * 30, // 30m + memoryCacheLifetime: 1000 * 60, // 1m + fetcher: (key) => this.userProfilesRepository.findOneByOrFail({ userId: key }), + toRedisConverter: (value) => JSON.stringify(value), + fromRedisConverter: (value) => JSON.parse(value), // TODO: date型の考慮 + }); + + this.userMutingsCache = new RedisKVCache<Set<string>>(this.redisClient, 'userMutings', { + lifetime: 1000 * 60 * 30, // 30m + memoryCacheLifetime: 1000 * 60, // 1m + fetcher: (key) => this.mutingsRepository.find({ where: { muterId: key }, select: ['muteeId'] }).then(xs => new Set(xs.map(x => x.muteeId))), + toRedisConverter: (value) => JSON.stringify(Array.from(value)), + fromRedisConverter: (value) => new Set(JSON.parse(value)), + }); + + this.userBlockingCache = new RedisKVCache<Set<string>>(this.redisClient, 'userBlocking', { + lifetime: 1000 * 60 * 30, // 30m + memoryCacheLifetime: 1000 * 60, // 1m + fetcher: (key) => this.blockingsRepository.find({ where: { blockerId: key }, select: ['blockeeId'] }).then(xs => new Set(xs.map(x => x.blockeeId))), + toRedisConverter: (value) => JSON.stringify(Array.from(value)), + fromRedisConverter: (value) => new Set(JSON.parse(value)), + }); + + this.userBlockedCache = new RedisKVCache<Set<string>>(this.redisClient, 'userBlocked', { + lifetime: 1000 * 60 * 30, // 30m + memoryCacheLifetime: 1000 * 60, // 1m + fetcher: (key) => this.blockingsRepository.find({ where: { blockeeId: key }, select: ['blockerId'] }).then(xs => new Set(xs.map(x => x.blockerId))), + toRedisConverter: (value) => JSON.stringify(Array.from(value)), + fromRedisConverter: (value) => new Set(JSON.parse(value)), + }); + + this.renoteMutingsCache = new RedisKVCache<Set<string>>(this.redisClient, 'renoteMutings', { + lifetime: 1000 * 60 * 30, // 30m + memoryCacheLifetime: 1000 * 60, // 1m + fetcher: (key) => this.renoteMutingsRepository.find({ where: { muterId: key }, select: ['muteeId'] }).then(xs => new Set(xs.map(x => x.muteeId))), + toRedisConverter: (value) => JSON.stringify(Array.from(value)), + fromRedisConverter: (value) => new Set(JSON.parse(value)), + }); + + this.userFollowingsCache = new RedisKVCache<Set<string>>(this.redisClient, 'userFollowings', { + lifetime: 1000 * 60 * 30, // 30m + memoryCacheLifetime: 1000 * 60, // 1m + fetcher: (key) => this.followingsRepository.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.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.redisSubscriber.on('message', this.onMessage); } diff --git a/packages/backend/src/core/DeleteAccountService.ts b/packages/backend/src/core/DeleteAccountService.ts index 2acb5f2303..327283106f 100644 --- a/packages/backend/src/core/DeleteAccountService.ts +++ b/packages/backend/src/core/DeleteAccountService.ts @@ -36,8 +36,5 @@ export class DeleteAccountService { await this.usersRepository.update(user.id, { isDeleted: true, }); - - // Terminate streaming - this.globalEventService.publishUserEvent(user.id, 'terminate', {}); } } diff --git a/packages/backend/src/core/GlobalEventService.ts b/packages/backend/src/core/GlobalEventService.ts index d261a6c657..25c064a2b4 100644 --- a/packages/backend/src/core/GlobalEventService.ts +++ b/packages/backend/src/core/GlobalEventService.ts @@ -14,7 +14,6 @@ import type { MainStreamTypes, NoteStreamTypes, UserListStreamTypes, - UserStreamTypes, } from '@/server/api/stream/types.js'; import type { Packed } from '@/misc/json-schema.js'; import { DI } from '@/di-symbols.js'; @@ -50,11 +49,6 @@ export class GlobalEventService { } @bindThis - public publishUserEvent<K extends keyof UserStreamTypes>(userId: User['id'], type: K, value?: UserStreamTypes[K]): void { - this.publish(`user:${userId}`, type, typeof value === 'undefined' ? null : value); - } - - @bindThis public publishBroadcastStream<K extends keyof BroadcastTypes>(type: K, value?: BroadcastTypes[K]): void { this.publish('broadcast', type, typeof value === 'undefined' ? null : value); } diff --git a/packages/backend/src/core/NotificationService.ts b/packages/backend/src/core/NotificationService.ts index 9c179f9318..366dc08c02 100644 --- a/packages/backend/src/core/NotificationService.ts +++ b/packages/backend/src/core/NotificationService.ts @@ -73,7 +73,7 @@ export class NotificationService implements OnApplicationShutdown { type: Notification['type'], data: Partial<Notification>, ): Promise<Notification | null> { - const profile = await this.cacheService.userProfileCache.fetch(notifieeId, () => this.userProfilesRepository.findOneByOrFail({ userId: notifieeId })); + const profile = await this.cacheService.userProfileCache.fetch(notifieeId); const isMuted = profile.mutingNotificationTypes.includes(type); if (isMuted) return null; @@ -82,8 +82,8 @@ export class NotificationService implements OnApplicationShutdown { return null; } - const mutings = await this.cacheService.userMutingsCache.fetch(notifieeId, () => this.mutingsRepository.findBy({ muterId: notifieeId }).then(xs => xs.map(x => x.muteeId))); - if (mutings.includes(data.notifierId)) { + const mutings = await this.cacheService.userMutingsCache.fetch(notifieeId); + if (mutings.has(data.notifierId)) { return null; } } diff --git a/packages/backend/src/core/UserBlockingService.ts b/packages/backend/src/core/UserBlockingService.ts index 040b6de2ef..6eaef8f68a 100644 --- a/packages/backend/src/core/UserBlockingService.ts +++ b/packages/backend/src/core/UserBlockingService.ts @@ -1,40 +1,26 @@ -import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common'; -import Redis from 'ioredis'; +import { Inject, Injectable } from '@nestjs/common'; import { IdService } from '@/core/IdService.js'; import type { User } from '@/models/entities/User.js'; import type { Blocking } from '@/models/entities/Blocking.js'; import { QueueService } from '@/core/QueueService.js'; import { GlobalEventService } from '@/core/GlobalEventService.js'; -import PerUserFollowingChart from '@/core/chart/charts/per-user-following.js'; import { DI } from '@/di-symbols.js'; -import type { UsersRepository, FollowingsRepository, FollowRequestsRepository, BlockingsRepository, UserListsRepository, UserListJoiningsRepository } from '@/models/index.js'; +import type { FollowRequestsRepository, BlockingsRepository, UserListsRepository, UserListJoiningsRepository } from '@/models/index.js'; import Logger from '@/logger.js'; import { UserEntityService } from '@/core/entities/UserEntityService.js'; import { ApRendererService } from '@/core/activitypub/ApRendererService.js'; import { LoggerService } from '@/core/LoggerService.js'; import { WebhookService } from '@/core/WebhookService.js'; import { bindThis } from '@/decorators.js'; -import { MemoryKVCache } from '@/misc/cache.js'; -import { StreamMessages } from '@/server/api/stream/types.js'; +import { CacheService } from '@/core/CacheService.js'; +import { UserFollowingService } from '@/core/UserFollowingService.js'; @Injectable() -export class UserBlockingService implements OnApplicationShutdown { +export class UserBlockingService { private logger: Logger; - // キーがユーザーIDで、値がそのユーザーがブロックしているユーザーのIDのリストなキャッシュ - private blockingsByUserIdCache: MemoryKVCache<User['id'][]>; - constructor( - @Inject(DI.redisSubscriber) - private redisSubscriber: Redis.Redis, - - @Inject(DI.usersRepository) - private usersRepository: UsersRepository, - - @Inject(DI.followingsRepository) - private followingsRepository: FollowingsRepository, - @Inject(DI.followRequestsRepository) private followRequestsRepository: FollowRequestsRepository, @@ -47,47 +33,17 @@ export class UserBlockingService implements OnApplicationShutdown { @Inject(DI.userListJoiningsRepository) private userListJoiningsRepository: UserListJoiningsRepository, + private cacheService: CacheService, + private userFollowingService: UserFollowingService, private userEntityService: UserEntityService, private idService: IdService, private queueService: QueueService, private globalEventService: GlobalEventService, private webhookService: WebhookService, private apRendererService: ApRendererService, - private perUserFollowingChart: PerUserFollowingChart, private loggerService: LoggerService, ) { this.logger = this.loggerService.getLogger('user-block'); - - this.blockingsByUserIdCache = new MemoryKVCache<User['id'][]>(Infinity); - - this.redisSubscriber.on('message', this.onMessage); - } - - @bindThis - private async onMessage(_: string, data: string): Promise<void> { - const obj = JSON.parse(data); - - if (obj.channel === 'internal') { - const { type, body } = obj.message as StreamMessages['internal']['payload']; - switch (type) { - case 'blockingCreated': { - const cached = this.blockingsByUserIdCache.get(body.blockerId); - if (cached) { - this.blockingsByUserIdCache.set(body.blockerId, [...cached, ...[body.blockeeId]]); - } - break; - } - case 'blockingDeleted': { - const cached = this.blockingsByUserIdCache.get(body.blockerId); - if (cached) { - this.blockingsByUserIdCache.set(body.blockerId, cached.filter(x => x !== body.blockeeId)); - } - break; - } - default: - break; - } - } } @bindThis @@ -95,8 +51,8 @@ export class UserBlockingService implements OnApplicationShutdown { await Promise.all([ this.cancelRequest(blocker, blockee), this.cancelRequest(blockee, blocker), - this.unFollow(blocker, blockee), - this.unFollow(blockee, blocker), + this.userFollowingService.unfollow(blocker, blockee), + this.userFollowingService.unfollow(blockee, blocker), this.removeFromList(blockee, blocker), ]); @@ -111,6 +67,9 @@ export class UserBlockingService implements OnApplicationShutdown { await this.blockingsRepository.insert(blocking); + this.cacheService.userBlockingCache.refresh(blocker.id); + this.cacheService.userBlockedCache.refresh(blockee.id); + this.globalEventService.publishInternalEvent('blockingCreated', { blockerId: blocker.id, blockeeId: blockee.id, @@ -148,7 +107,6 @@ export class UserBlockingService implements OnApplicationShutdown { this.userEntityService.pack(followee, follower, { detail: true, }).then(async packed => { - this.globalEventService.publishUserEvent(follower.id, 'unfollow', packed); this.globalEventService.publishMainStream(follower.id, 'unfollow', packed); const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow')); @@ -174,54 +132,6 @@ export class UserBlockingService implements OnApplicationShutdown { } @bindThis - private async unFollow(follower: User, followee: User) { - const following = await this.followingsRepository.findOneBy({ - followerId: follower.id, - followeeId: followee.id, - }); - - if (following == null) { - return; - } - - await Promise.all([ - this.followingsRepository.delete(following.id), - this.usersRepository.decrement({ id: follower.id }, 'followingCount', 1), - this.usersRepository.decrement({ id: followee.id }, 'followersCount', 1), - this.perUserFollowingChart.update(follower, followee, false), - ]); - - // Publish unfollow event - if (this.userEntityService.isLocalUser(follower)) { - this.userEntityService.pack(followee, follower, { - detail: true, - }).then(async packed => { - this.globalEventService.publishUserEvent(follower.id, 'unfollow', packed); - this.globalEventService.publishMainStream(follower.id, 'unfollow', packed); - - const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow')); - for (const webhook of webhooks) { - this.queueService.webhookDeliver(webhook, 'unfollow', { - user: packed, - }); - } - }); - } - - // リモートにフォローをしていたらUndoFollow送信 - if (this.userEntityService.isLocalUser(follower) && this.userEntityService.isRemoteUser(followee)) { - const content = this.apRendererService.addContext(this.apRendererService.renderUndo(this.apRendererService.renderFollow(follower, followee), follower)); - this.queueService.deliver(follower, content, followee.inbox, false); - } - - // リモートからフォローをされていたらRejectFollow送信 - if (this.userEntityService.isLocalUser(followee) && this.userEntityService.isRemoteUser(follower)) { - const content = this.apRendererService.addContext(this.apRendererService.renderReject(this.apRendererService.renderFollow(follower, followee), followee)); - this.queueService.deliver(followee, content, follower.inbox, false); - } - } - - @bindThis private async removeFromList(listOwner: User, user: User) { const userLists = await this.userListsRepository.findBy({ userId: listOwner.id, @@ -254,6 +164,9 @@ export class UserBlockingService implements OnApplicationShutdown { await this.blockingsRepository.delete(blocking.id); + this.cacheService.userBlockingCache.refresh(blocker.id); + this.cacheService.userBlockedCache.refresh(blockee.id); + this.globalEventService.publishInternalEvent('blockingDeleted', { blockerId: blocker.id, blockeeId: blockee.id, @@ -268,17 +181,6 @@ export class UserBlockingService implements OnApplicationShutdown { @bindThis public async checkBlocked(blockerId: User['id'], blockeeId: User['id']): Promise<boolean> { - const blockedUserIds = await this.blockingsByUserIdCache.fetch(blockerId, () => this.blockingsRepository.find({ - where: { - blockerId, - }, - select: ['blockeeId'], - }).then(records => records.map(record => record.blockeeId))); - return blockedUserIds.includes(blockeeId); - } - - @bindThis - public onApplicationShutdown(signal?: string | undefined) { - this.redisSubscriber.off('message', this.onMessage); + return (await this.cacheService.userBlockingCache.fetch(blockerId)).has(blockeeId); } } diff --git a/packages/backend/src/core/UserFollowingService.ts b/packages/backend/src/core/UserFollowingService.ts index b51b553c70..4f22c5cd46 100644 --- a/packages/backend/src/core/UserFollowingService.ts +++ b/packages/backend/src/core/UserFollowingService.ts @@ -18,6 +18,7 @@ import { ApRendererService } from '@/core/activitypub/ApRendererService.js'; import { bindThis } from '@/decorators.js'; import { UserBlockingService } from '@/core/UserBlockingService.js'; import { MetaService } from '@/core/MetaService.js'; +import { CacheService } from '@/core/CacheService.js'; import Logger from '../logger.js'; const logger = new Logger('following/create'); @@ -53,6 +54,7 @@ export class UserFollowingService { @Inject(DI.instancesRepository) private instancesRepository: InstancesRepository, + private cacheService: CacheService, private userEntityService: UserEntityService, private userBlockingService: UserBlockingService, private idService: IdService, @@ -172,6 +174,8 @@ export class UserFollowingService { } }); + this.cacheService.userFollowingsCache.refresh(follower.id); + const req = await this.followRequestsRepository.findOneBy({ followeeId: followee.id, followerId: follower.id, @@ -225,7 +229,6 @@ export class UserFollowingService { this.userEntityService.pack(followee.id, follower, { detail: true, }).then(async packed => { - this.globalEventService.publishUserEvent(follower.id, 'follow', packed as Packed<'UserDetailedNotMe'>); this.globalEventService.publishMainStream(follower.id, 'follow', packed as Packed<'UserDetailedNotMe'>); const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('follow')); @@ -279,6 +282,8 @@ export class UserFollowingService { await this.followingsRepository.delete(following.id); + this.cacheService.userFollowingsCache.refresh(follower.id); + this.decrementFollowing(follower, followee); // Publish unfollow event @@ -286,7 +291,6 @@ export class UserFollowingService { this.userEntityService.pack(followee.id, follower, { detail: true, }).then(async packed => { - this.globalEventService.publishUserEvent(follower.id, 'unfollow', packed); this.globalEventService.publishMainStream(follower.id, 'unfollow', packed); const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow')); @@ -579,7 +583,6 @@ export class UserFollowingService { detail: true, }); - this.globalEventService.publishUserEvent(follower.id, 'unfollow', packedFollowee); this.globalEventService.publishMainStream(follower.id, 'unfollow', packedFollowee); const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow')); diff --git a/packages/backend/src/core/UserMutingService.ts b/packages/backend/src/core/UserMutingService.ts index e98f11709f..657b5764b9 100644 --- a/packages/backend/src/core/UserMutingService.ts +++ b/packages/backend/src/core/UserMutingService.ts @@ -1,34 +1,47 @@ import { Inject, Injectable } from '@nestjs/common'; -import type { UsersRepository, MutingsRepository } from '@/models/index.js'; +import { In } from 'typeorm'; +import type { MutingsRepository, Muting } from '@/models/index.js'; import { IdService } from '@/core/IdService.js'; -import { QueueService } from '@/core/QueueService.js'; -import { GlobalEventService } from '@/core/GlobalEventService.js'; import type { User } from '@/models/entities/User.js'; import { DI } from '@/di-symbols.js'; import { bindThis } from '@/decorators.js'; +import { CacheService } from '@/core/CacheService'; @Injectable() export class UserMutingService { constructor( - @Inject(DI.usersRepository) - private usersRepository: UsersRepository, - @Inject(DI.mutingsRepository) private mutingsRepository: MutingsRepository, private idService: IdService, - private queueService: QueueService, - private globalEventService: GlobalEventService, + private cacheService: CacheService, ) { } @bindThis - public async mute(user: User, target: User): Promise<void> { + public async mute(user: User, target: User, expiresAt: Date | null = null): Promise<void> { await this.mutingsRepository.insert({ id: this.idService.genId(), createdAt: new Date(), + expiresAt: expiresAt ?? null, muterId: user.id, muteeId: target.id, }); + + this.cacheService.userMutingsCache.refresh(user.id); + } + + @bindThis + public async unmute(mutings: Muting[]): Promise<void> { + if (mutings.length === 0) return; + + await this.mutingsRepository.delete({ + id: In(mutings.map(m => m.id)), + }); + + const muterIds = [...new Set(mutings.map(m => m.muterId))]; + for (const muterId of muterIds) { + this.cacheService.userMutingsCache.refresh(muterId); + } } } |