From 6816bce88cb1f2939aaa1801af5951c61e66ba2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=81=BE=E3=81=A3=E3=81=A1=E3=82=83=E3=81=A8=E3=83=BC?= =?UTF-8?q?=E3=81=AB=E3=82=85?= <17376330+u1-liquid@users.noreply.github.com> Date: Fri, 5 Apr 2024 14:06:14 +0900 Subject: [PATCH] =?UTF-8?q?fix(backend/UserSuspendService):=20=E5=87=8D?= =?UTF-8?q?=E7=B5=90=E3=83=BB=E8=A7=A3=E5=87=8D=E3=81=AE=E5=87=A6=E7=90=86?= =?UTF-8?q?=E3=81=A7following=E3=83=86=E3=83=BC=E3=83=96=E3=83=AB=E3=81=AE?= =?UTF-8?q?=E5=85=A8=E3=81=A6=E3=81=AE=E3=83=87=E3=83=BC=E3=82=BF=E3=82=92?= =?UTF-8?q?fetch=E3=81=97=E3=81=A6=E3=81=97=E3=81=BE=E3=81=84OOM=E3=81=AB?= =?UTF-8?q?=E3=81=AA=E3=82=8B=E5=95=8F=E9=A1=8C=E3=82=92=E4=BF=AE=E6=AD=A3?= =?UTF-8?q?=20(MisskeyIO#598)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend/src/core/UserSuspendService.ts | 115 ++++++++++++------ 1 file changed, 78 insertions(+), 37 deletions(-) diff --git a/packages/backend/src/core/UserSuspendService.ts b/packages/backend/src/core/UserSuspendService.ts index d594a223f4..cc8540f0ae 100644 --- a/packages/backend/src/core/UserSuspendService.ts +++ b/packages/backend/src/core/UserSuspendService.ts @@ -4,86 +4,127 @@ */ import { Inject, Injectable } from '@nestjs/common'; -import { Not, IsNull } from 'typeorm'; -import type { FollowingsRepository } from '@/models/_.js'; +import { Not, IsNull, Brackets, DataSource, EntityManager } from 'typeorm'; +import { bindThis } from '@/decorators.js'; +import { DI } from '@/di-symbols.js'; +import type Logger from '@/logger.js'; import type { MiUser } from '@/models/User.js'; +import type { FollowingsRepository } from '@/models/_.js'; import { QueueService } from '@/core/QueueService.js'; import { GlobalEventService } from '@/core/GlobalEventService.js'; -import { DI } from '@/di-symbols.js'; import { ApRendererService } from '@/core/activitypub/ApRendererService.js'; import { UserEntityService } from '@/core/entities/UserEntityService.js'; -import { bindThis } from '@/decorators.js'; +import { LoggerService } from '@/core/LoggerService.js'; @Injectable() export class UserSuspendService { + public logger: Logger; + constructor( + @Inject(DI.db) + private db: DataSource, + @Inject(DI.followingsRepository) private followingsRepository: FollowingsRepository, - private userEntityService: UserEntityService, private queueService: QueueService, private globalEventService: GlobalEventService, private apRendererService: ApRendererService, + private userEntityService: UserEntityService, + private loggerService: LoggerService, ) { + this.logger = this.loggerService.getLogger('account:suspend'); } @bindThis public async doPostSuspend(user: { id: MiUser['id']; host: MiUser['host'] }): Promise { + this.logger.warn(`doPostSuspend: ${user.id} (host: ${user.host})`); + this.globalEventService.publishInternalEvent('userChangeSuspendedState', { id: user.id, isSuspended: true }); if (this.userEntityService.isLocalUser(user)) { // 知り得る全SharedInboxにDelete配信 const content = this.apRendererService.addContext(this.apRendererService.renderDelete(this.userEntityService.genLocalUserUri(user.id), user)); - const queue: string[] = []; + this.logger.info(`Delivering delete activity to all shared inboxes of ${user.id}`); + await this.db.transaction(async (transactionalEntityManager: EntityManager) => { + const inboxesCte = transactionalEntityManager.createQueryBuilder() + .select('distinct coalesce(following.followerSharedInbox, following.followeeSharedInbox)', 'inbox') + .from(this.followingsRepository.metadata.targetName, 'following') + .where(new Brackets((qb) => qb.where({ followerHost: Not(IsNull()) }).orWhere({ followeeHost: Not(IsNull()) }))) + .andWhere(new Brackets((qb) => qb.where({ followerSharedInbox: Not(IsNull()) }).orWhere({ followeeSharedInbox: Not(IsNull()) }))) + .orderBy('inbox'); - const followings = await this.followingsRepository.find({ - where: [ - { followerSharedInbox: Not(IsNull()) }, - { followeeSharedInbox: Not(IsNull()) }, - ], - select: ['followerSharedInbox', 'followeeSharedInbox'], + let offset = 0; + let cursor = ''; + while (true) { // eslint-disable-line @typescript-eslint/no-unnecessary-condition, no-constant-condition + const inboxes = await transactionalEntityManager.createQueryBuilder().addCommonTableExpression(inboxesCte, 'inboxes') + .select('inbox').from('inboxes', 'inboxes').where('inbox > :cursor', { cursor }).limit(500).getRawMany<{ inbox: string }>() + .then((rows) => rows.map((row) => row.inbox)); + + if (inboxes.length === 0) break; + + this.logger.info(`Delivering delete activity to ${offset} - ${offset + inboxes.length} shared inboxes of ${user.id}`); + for (const inbox of inboxes) { + try { + this.queueService.deliver(user, content, inbox, true); + } catch (err) { + this.logger.error(`Failed to deliver delete activity to ${inbox}: ${err}`, { error: err, inbox }); + } + } + + offset += inboxes.length; + cursor = inboxes[inboxes.length - 1]; + } }); - const inboxes = followings.map(x => x.followerSharedInbox ?? x.followeeSharedInbox); - - for (const inbox of inboxes) { - if (inbox != null && !queue.includes(inbox)) queue.push(inbox); - } - - for (const inbox of queue) { - this.queueService.deliver(user, content, inbox, true); - } + this.logger.info(`Scheduled delete activity delivery to all shared inboxes of ${user.id}`); } } @bindThis public async doPostUnsuspend(user: MiUser): Promise { + this.logger.warn(`doPostUnsuspend: ${user.id}`); + this.globalEventService.publishInternalEvent('userChangeSuspendedState', { id: user.id, isSuspended: false }); if (this.userEntityService.isLocalUser(user)) { // 知り得る全SharedInboxにUndo Delete配信 const content = this.apRendererService.addContext(this.apRendererService.renderUndo(this.apRendererService.renderDelete(this.userEntityService.genLocalUserUri(user.id), user), user)); - const queue: string[] = []; + this.logger.info(`Delivering undo delete activity to all shared inboxes of ${user.id}`); + await this.db.transaction(async (transactionalEntityManager: EntityManager) => { + const inboxesCte = transactionalEntityManager.createQueryBuilder() + .select('distinct coalesce(following.followerSharedInbox, following.followeeSharedInbox)', 'inbox') + .from(this.followingsRepository.metadata.targetName, 'following') + .where(new Brackets((qb) => qb.where({ followerHost: Not(IsNull()) }).orWhere({ followeeHost: Not(IsNull()) }))) + .andWhere(new Brackets((qb) => qb.where({ followerSharedInbox: Not(IsNull()) }).orWhere({ followeeSharedInbox: Not(IsNull()) }))) + .orderBy('inbox'); - const followings = await this.followingsRepository.find({ - where: [ - { followerSharedInbox: Not(IsNull()) }, - { followeeSharedInbox: Not(IsNull()) }, - ], - select: ['followerSharedInbox', 'followeeSharedInbox'], + let offset = 0; + let cursor = ''; + while (true) { // eslint-disable-line @typescript-eslint/no-unnecessary-condition, no-constant-condition + const inboxes = await transactionalEntityManager.createQueryBuilder().addCommonTableExpression(inboxesCte, 'inboxes') + .select('inbox').from('inboxes', 'inboxes').where('inbox > :cursor', { cursor }).limit(500).getRawMany<{ inbox: string }>() + .then((rows) => rows.map((row) => row.inbox)); + + if (inboxes.length === 0) break; + + this.logger.info(`Delivering undo delete activity to ${offset} - ${offset + inboxes.length} shared inboxes of ${user.id}`); + for (const inbox of inboxes) { + try { + this.queueService.deliver(user, content, inbox, true); + } catch (err) { + this.logger.error(`Failed to deliver undo delete activity to ${inbox}: ${err}`, { error: err, inbox }); + } + } + + offset += inboxes.length; + cursor = inboxes[inboxes.length - 1]; + } }); - const inboxes = followings.map(x => x.followerSharedInbox ?? x.followeeSharedInbox); - - for (const inbox of inboxes) { - if (inbox != null && !queue.includes(inbox)) queue.push(inbox); - } - - for (const inbox of queue) { - this.queueService.deliver(user as any, content, inbox, true); - } + this.logger.info(`Scheduled undo delete activity delivery to all shared inboxes of ${user.id}`); } } }