feat: queueing bulk follow/unfollow and block/unblock (#10544)

* wrap follow/unfollow and block/unblock as job queue

* create import job to follow in each iteration

* make relationship jobs concurrent

* replace to job queue if called repeatedly

* use addBulk to import

* omit stream when importing

* fix job caller

* use ThinUser instead of User to reduce redis memory consumption

* createImportFollowingToDbJobの呼び出し方を変える, 型補強

* Force ThinUser

* オブジェクト操作のみのメソッド名はgenerate...Data

* Force ThinUser in generateRelationshipJobData

* silent bulk unfollow at admin api endpoint

---------

Co-authored-by: tamaina <tamaina@hotmail.co.jp>

1 files changed, 68 insertions, 0 deletions

diff --git a/packages/backend/src/queue/processors/RelationshipProcessorService.ts b/packages/backend/src/queue/processors/RelationshipProcessorService.ts
new file mode 100644
index 0000000000..a5006dcf03
--- /dev/null
+++ b/packages/backend/src/queue/processors/RelationshipProcessorService.ts
@@ -0,0 +1,68 @@
+import { Inject, Injectable } from '@nestjs/common';
+import type Bull from 'bull';
+
+import { UserFollowingService } from '@/core/UserFollowingService.js';
+import { UserBlockingService } from '@/core/UserBlockingService.js';
+import { bindThis } from '@/decorators.js';
+import type Logger from '@/logger.js';
+
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import { RelationshipJobData } from '../types.js';
+import type { UsersRepository } from '@/models/index.js';
+import { DI } from '@/di-symbols.js';
+
+@Injectable()
+export class RelationshipProcessorService {
+	private logger: Logger;
+
+	constructor(
+		@Inject(DI.usersRepository)
+		private usersRepository: UsersRepository,
+
+		private queueLoggerService: QueueLoggerService,
+		private userFollowingService: UserFollowingService,
+		private userBlockingService: UserBlockingService,
+	) {
+		this.logger = this.queueLoggerService.logger.createSubLogger('follow-block');
+	}
+
+	@bindThis
+	public async processFollow(job: Bull.Job<RelationshipJobData>): Promise<string> {
+		this.logger.info(`${job.data.from.id} is trying to follow ${job.data.to.id}`);
+		await this.userFollowingService.follow(job.data.from, job.data.to, job.data.requestId, job.data.silent);
+		return 'ok';
+	}
+
+	@bindThis
+	public async processUnfollow(job: Bull.Job<RelationshipJobData>): Promise<string> {
+		this.logger.info(`${job.data.from.id} is trying to unfollow ${job.data.to.id}`);
+		const [follower, followee] = await Promise.all([
+			this.usersRepository.findOneByOrFail({ id: job.data.from.id }),
+			this.usersRepository.findOneByOrFail({ id: job.data.to.id }),
+		]);
+		await this.userFollowingService.unfollow(follower, followee, job.data.silent);
+		return 'ok';
+	}
+
+	@bindThis
+	public async processBlock(job: Bull.Job<RelationshipJobData>): Promise<string> {
+		this.logger.info(`${job.data.from.id} is trying to block ${job.data.to.id}`);
+		const [blockee, blocker] = await Promise.all([
+			this.usersRepository.findOneByOrFail({ id: job.data.from.id }),
+			this.usersRepository.findOneByOrFail({ id: job.data.to.id }),
+		]);
+		await this.userBlockingService.block(blockee, blocker, job.data.silent);
+		return 'ok';
+	}
+
+	@bindThis
+	public async processUnblock(job: Bull.Job<RelationshipJobData>): Promise<string> {
+		this.logger.info(`${job.data.from.id} is trying to unblock ${job.data.to.id}`);
+		const [blockee, blocker] = await Promise.all([
+			this.usersRepository.findOneByOrFail({ id: job.data.from.id }),
+			this.usersRepository.findOneByOrFail({ id: job.data.to.id }),
+		]);
+		await this.userBlockingService.unblock(blockee, blocker);
+		return 'ok';
+	}
+}