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, 11 insertions, 2 deletions

diff --git a/packages/backend/src/core/QueueModule.ts b/packages/backend/src/core/QueueModule.ts
index 8733a7d7eb..bac85d7a15 100644
--- a/packages/backend/src/core/QueueModule.ts
+++ b/packages/backend/src/core/QueueModule.ts
@@ -3,7 +3,7 @@ import Bull from 'bull';
 import { DI } from '@/di-symbols.js';
 import type { Config } from '@/config.js';
 import type { Provider } from '@nestjs/common';
-import type { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData, WebhookDeliverJobData } from '../queue/types.js';
+import type { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData, WebhookDeliverJobData, RelationshipJobData, DbJobMap } from '../queue/types.js';
 
 function q<T>(config: Config, name: string, limitPerSec = -1) {
 	return new Bull<T>(name, {
@@ -41,7 +41,8 @@ export type SystemQueue = Bull.Queue<Record<string, unknown>>;
 export type EndedPollNotificationQueue = Bull.Queue<EndedPollNotificationJobData>;
 export type DeliverQueue = Bull.Queue<DeliverJobData>;
 export type InboxQueue = Bull.Queue<InboxJobData>;
-export type DbQueue = Bull.Queue<DbJobData>;
+export type DbQueue = Bull.Queue<DbJobData<keyof DbJobMap>>;
+export type RelationshipQueue = Bull.Queue<RelationshipJobData>;
 export type ObjectStorageQueue = Bull.Queue<ObjectStorageJobData>;
 export type WebhookDeliverQueue = Bull.Queue<WebhookDeliverJobData>;
 
@@ -75,6 +76,12 @@ const $db: Provider = {
 	inject: [DI.config],
 };
 
+const $relationship: Provider = {
+	provide: 'queue:relationship',
+	useFactory: (config: Config) => q(config, 'relationship'),
+	inject: [DI.config],
+};
+
 const $objectStorage: Provider = {
 	provide: 'queue:objectStorage',
 	useFactory: (config: Config) => q(config, 'objectStorage'),
@@ -96,6 +103,7 @@ const $webhookDeliver: Provider = {
 		$deliver,
 		$inbox,
 		$db,
+		$relationship,
 		$objectStorage,
 		$webhookDeliver,
 	],
@@ -105,6 +113,7 @@ const $webhookDeliver: Provider = {
 		$deliver,
 		$inbox,
 		$db,
+		$relationship,
 		$objectStorage,
 		$webhookDeliver,
 	],