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>

packages/backend/src/queue/DbQueueProcessorsService.ts

@@ -52,8 +52,10 @@ export class DbQueueProcessorsService {
 		q.process('exportBlocking', (job, done) => this.exportBlockingProcessorService.process(job, done));
 		q.process('exportUserLists', (job, done) => this.exportUserListsProcessorService.process(job, done));
 		q.process('importFollowing', (job, done) => this.importFollowingProcessorService.process(job, done));
+		q.process('importFollowingToDb', (job) => this.importFollowingProcessorService.processDb(job));
 		q.process('importMuting', (job, done) => this.importMutingProcessorService.process(job, done));
 		q.process('importBlocking', (job, done) => this.importBlockingProcessorService.process(job, done));
+		q.process('importBlockingToDb', (job) => this.importBlockingProcessorService.processDb(job));
 		q.process('importUserLists', (job, done) => this.importUserListsProcessorService.process(job, done));
 		q.process('importCustomEmojis', (job, done) => this.importCustomEmojisProcessorService.process(job, done));
 		q.process('deleteAccount', (job) => this.deleteAccountProcessorService.process(job));

packages/backend/src/queue/QueueProcessorModule.ts

@@ -4,6 +4,7 @@ import { GlobalModule } from '@/GlobalModule.js';
 import { QueueLoggerService } from './QueueLoggerService.js';
 import { QueueProcessorService } from './QueueProcessorService.js';
 import { DbQueueProcessorsService } from './DbQueueProcessorsService.js';
+import { RelationshipQueueProcessorsService } from './RelationshipQueueProcessorsService.js';
 import { ObjectStorageQueueProcessorsService } from './ObjectStorageQueueProcessorsService.js';
 import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
 import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
@@ -32,6 +33,7 @@ import { ResyncChartsProcessorService } from './processors/ResyncChartsProcessor
 import { TickChartsProcessorService } from './processors/TickChartsProcessorService.js';
 import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js';
 import { ExportFavoritesProcessorService } from './processors/ExportFavoritesProcessorService.js';
+import { RelationshipProcessorService } from './processors/RelationshipProcessorService.js';
 
 @Module({
 	imports: [
@@ -61,9 +63,11 @@ import { ExportFavoritesProcessorService } from './processors/ExportFavoritesPro
 		DeleteAccountProcessorService,
 		DeleteFileProcessorService,
 		CleanRemoteFilesProcessorService,
+		RelationshipProcessorService,
 		SystemQueueProcessorsService,
 		ObjectStorageQueueProcessorsService,
 		DbQueueProcessorsService,
+		RelationshipQueueProcessorsService,
 		WebhookDeliverProcessorService,
 		EndedPollNotificationProcessorService,
 		DeliverProcessorService,

packages/backend/src/queue/QueueProcessorService.ts

@@ -13,6 +13,7 @@ import { EndedPollNotificationProcessorService } from './processors/EndedPollNot
 import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
 import { InboxProcessorService } from './processors/InboxProcessorService.js';
 import { QueueLoggerService } from './QueueLoggerService.js';
+import { RelationshipQueueProcessorsService } from './RelationshipQueueProcessorsService.js';
 
 @Injectable()
 export class QueueProcessorService {
@@ -27,6 +28,7 @@ export class QueueProcessorService {
 		private systemQueueProcessorsService: SystemQueueProcessorsService,
 		private objectStorageQueueProcessorsService: ObjectStorageQueueProcessorsService,
 		private dbQueueProcessorsService: DbQueueProcessorsService,
+		private relationshipQueueProcessorsService: RelationshipQueueProcessorsService,
 		private webhookDeliverProcessorService: WebhookDeliverProcessorService,
 		private endedPollNotificationProcessorService: EndedPollNotificationProcessorService,
 		private deliverProcessorService: DeliverProcessorService,
@@ -52,14 +54,15 @@ export class QueueProcessorService {
 				};
 			}
 		}
-	
+
 		const systemLogger = this.logger.createSubLogger('system');
 		const deliverLogger = this.logger.createSubLogger('deliver');
 		const webhookLogger = this.logger.createSubLogger('webhook');
 		const inboxLogger = this.logger.createSubLogger('inbox');
 		const dbLogger = this.logger.createSubLogger('db');
+		const relationshipLogger = this.logger.createSubLogger('relationship');
 		const objectStorageLogger = this.logger.createSubLogger('objectStorage');
-		
+
 		this.queueService.systemQueue
 			.on('waiting', (jobId) => systemLogger.debug(`waiting id=${jobId}`))
 			.on('active', (job) => systemLogger.debug(`active id=${job.id}`))
@@ -67,7 +70,7 @@ export class QueueProcessorService {
 			.on('failed', (job, err) => systemLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
 			.on('error', (job: any, err: Error) => systemLogger.error(`error ${err}`, { job, e: renderError(err) }))
 			.on('stalled', (job) => systemLogger.warn(`stalled id=${job.id}`));
-		
+
 		this.queueService.deliverQueue
 			.on('waiting', (jobId) => deliverLogger.debug(`waiting id=${jobId}`))
 			.on('active', (job) => deliverLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
@@ -75,7 +78,7 @@ export class QueueProcessorService {
 			.on('failed', (job, err) => deliverLogger.warn(`failed(${err}) ${getJobInfo(job)} to=${job.data.to}`))
 			.on('error', (job: any, err: Error) => deliverLogger.error(`error ${err}`, { job, e: renderError(err) }))
 			.on('stalled', (job) => deliverLogger.warn(`stalled ${getJobInfo(job)} to=${job.data.to}`));
-		
+
 		this.queueService.inboxQueue
 			.on('waiting', (jobId) => inboxLogger.debug(`waiting id=${jobId}`))
 			.on('active', (job) => inboxLogger.debug(`active ${getJobInfo(job, true)}`))
@@ -83,7 +86,7 @@ export class QueueProcessorService {
 			.on('failed', (job, err) => inboxLogger.warn(`failed(${err}) ${getJobInfo(job)} activity=${job.data.activity ? job.data.activity.id : 'none'}`, { job, e: renderError(err) }))
 			.on('error', (job: any, err: Error) => inboxLogger.error(`error ${err}`, { job, e: renderError(err) }))
 			.on('stalled', (job) => inboxLogger.warn(`stalled ${getJobInfo(job)} activity=${job.data.activity ? job.data.activity.id : 'none'}`));
-		
+
 		this.queueService.dbQueue
 			.on('waiting', (jobId) => dbLogger.debug(`waiting id=${jobId}`))
 			.on('active', (job) => dbLogger.debug(`active id=${job.id}`))
@@ -91,7 +94,15 @@ export class QueueProcessorService {
 			.on('failed', (job, err) => dbLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
 			.on('error', (job: any, err: Error) => dbLogger.error(`error ${err}`, { job, e: renderError(err) }))
 			.on('stalled', (job) => dbLogger.warn(`stalled id=${job.id}`));
-		
+
+		this.queueService.relationshipQueue
+			.on('waiting', (jobId) => relationshipLogger.debug(`waiting id=${jobId}`))
+			.on('active', (job) => relationshipLogger.debug(`active id=${job.id}`))
+			.on('completed', (job, result) => relationshipLogger.debug(`completed(${result}) id=${job.id}`))
+			.on('failed', (job, err) => relationshipLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
+			.on('error', (job: any, err: Error) => relationshipLogger.error(`error ${err}`, { job, e: renderError(err) }))
+			.on('stalled', (job) => relationshipLogger.warn(`stalled id=${job.id}`));
+
 		this.queueService.objectStorageQueue
 			.on('waiting', (jobId) => objectStorageLogger.debug(`waiting id=${jobId}`))
 			.on('active', (job) => objectStorageLogger.debug(`active id=${job.id}`))
@@ -99,7 +110,7 @@ export class QueueProcessorService {
 			.on('failed', (job, err) => objectStorageLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
 			.on('error', (job: any, err: Error) => objectStorageLogger.error(`error ${err}`, { job, e: renderError(err) }))
 			.on('stalled', (job) => objectStorageLogger.warn(`stalled id=${job.id}`));
-		
+
 		this.queueService.webhookDeliverQueue
 			.on('waiting', (jobId) => webhookLogger.debug(`waiting id=${jobId}`))
 			.on('active', (job) => webhookLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
@@ -107,26 +118,27 @@ export class QueueProcessorService {
 			.on('failed', (job, err) => webhookLogger.warn(`failed(${err}) ${getJobInfo(job)} to=${job.data.to}`))
 			.on('error', (job: any, err: Error) => webhookLogger.error(`error ${err}`, { job, e: renderError(err) }))
 			.on('stalled', (job) => webhookLogger.warn(`stalled ${getJobInfo(job)} to=${job.data.to}`));
-	
+
 		this.queueService.deliverQueue.process(this.config.deliverJobConcurrency ?? 128, (job) => this.deliverProcessorService.process(job));
 		this.queueService.inboxQueue.process(this.config.inboxJobConcurrency ?? 16, (job) => this.inboxProcessorService.process(job));
 		this.queueService.endedPollNotificationQueue.process((job, done) => this.endedPollNotificationProcessorService.process(job, done));
 		this.queueService.webhookDeliverQueue.process(64, (job) => this.webhookDeliverProcessorService.process(job));
 		this.dbQueueProcessorsService.start(this.queueService.dbQueue);
+		this.relationshipQueueProcessorsService.start(this.queueService.relationshipQueue);
 		this.objectStorageQueueProcessorsService.start(this.queueService.objectStorageQueue);
-	
+
 		this.queueService.systemQueue.add('tickCharts', {
 		}, {
 			repeat: { cron: '55 * * * *' },
 			removeOnComplete: true,
 		});
-	
+
 		this.queueService.systemQueue.add('resyncCharts', {
 		}, {
 			repeat: { cron: '0 0 * * *' },
 			removeOnComplete: true,
 		});
-	
+
 		this.queueService.systemQueue.add('cleanCharts', {
 		}, {
 			repeat: { cron: '0 0 * * *' },
@@ -138,19 +150,19 @@ export class QueueProcessorService {
 			repeat: { cron: '0 0 * * *' },
 			removeOnComplete: true,
 		});
-	
+
 		this.queueService.systemQueue.add('clean', {
 		}, {
 			repeat: { cron: '0 0 * * *' },
 			removeOnComplete: true,
 		});
-	
+
 		this.queueService.systemQueue.add('checkExpiredMutings', {
 		}, {
 			repeat: { cron: '*/5 * * * *' },
 			removeOnComplete: true,
 		});
-	
+
 		this.systemQueueProcessorsService.start(this.queueService.systemQueue);
 	}
 }

packages/backend/src/queue/RelationshipQueueProcessorsService.ts

@@ -0,0 +1,26 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { bindThis } from '@/decorators.js';
+import { RelationshipProcessorService } from './processors/RelationshipProcessorService.js';
+import type Bull from 'bull';
+import { DI } from '@/di-symbols.js';
+import type { Config } from '@/config.js';
+
+@Injectable()
+export class RelationshipQueueProcessorsService {
+	constructor(
+		@Inject(DI.config)
+		private config: Config,
+
+		private relationshipProcessorService: RelationshipProcessorService,
+	) {
+	}
+
+	@bindThis
+	public start(q: Bull.Queue): void {
+		const maxJobs = (this.config.deliverJobConcurrency ?? 128) / 4; // conservative?
+		q.process('follow', maxJobs, (job) => this.relationshipProcessorService.processFollow(job));
+		q.process('unfollow', maxJobs, (job) => this.relationshipProcessorService.processUnfollow(job));
+		q.process('block', maxJobs, (job) => this.relationshipProcessorService.processBlock(job));
+		q.process('unblock', maxJobs, (job) => this.relationshipProcessorService.processUnblock(job));
+	}
+}

packages/backend/src/queue/processors/DeleteDriveFilesProcessorService.ts

@@ -7,7 +7,7 @@ import type Logger from '@/logger.js';
 import { DriveService } from '@/core/DriveService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
 import { bindThis } from '@/decorators.js';
 
 @Injectable()
@@ -31,7 +31,7 @@ export class DeleteDriveFilesProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
 		this.logger.info(`Deleting drive files of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ExportBlockingProcessorService.ts

@@ -11,7 +11,7 @@ import { createTemp } from '@/misc/create-temp.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
 import { bindThis } from '@/decorators.js';
 
 @Injectable()
@@ -36,7 +36,7 @@ export class ExportBlockingProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
 		this.logger.info(`Exporting blocking of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ExportFavoritesProcessorService.ts

@@ -13,7 +13,7 @@ import type { Note } from '@/models/entities/Note.js';
 import { bindThis } from '@/decorators.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
 
 @Injectable()
 export class ExportFavoritesProcessorService {
@@ -42,7 +42,7 @@ export class ExportFavoritesProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
 		this.logger.info(`Exporting favorites of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ExportFollowingProcessorService.ts

@@ -12,7 +12,7 @@ import type { Following } from '@/models/entities/Following.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbExportFollowingData } from '../types.js';
 import { bindThis } from '@/decorators.js';
 
 @Injectable()
@@ -40,7 +40,7 @@ export class ExportFollowingProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbExportFollowingData>, done: () => void): Promise<void> {
 		this.logger.info(`Exporting following of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ExportMutingProcessorService.ts

@@ -11,7 +11,7 @@ import { createTemp } from '@/misc/create-temp.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
 import { bindThis } from '@/decorators.js';
 
 @Injectable()
@@ -39,7 +39,7 @@ export class ExportMutingProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
 		this.logger.info(`Exporting muting of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ExportNotesProcessorService.ts

@@ -13,7 +13,7 @@ import type { Note } from '@/models/entities/Note.js';
 import { bindThis } from '@/decorators.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
 
 @Injectable()
 export class ExportNotesProcessorService {
@@ -39,7 +39,7 @@ export class ExportNotesProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
 		this.logger.info(`Exporting notes of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ExportUserListsProcessorService.ts

@@ -11,7 +11,7 @@ import { createTemp } from '@/misc/create-temp.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
 import { bindThis } from '@/decorators.js';
 
 @Injectable()
@@ -39,7 +39,7 @@ export class ExportUserListsProcessorService {
 	}
 
 	@bindThis
-	public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+	public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
 		this.logger.info(`Exporting user lists of ${job.data.user.id} ...`);
 
 		const user = await this.usersRepository.findOneBy({ id: job.data.user.id });

packages/backend/src/queue/processors/ImportBlockingProcessorService.ts

@@ -1,38 +1,31 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { IsNull } from 'typeorm';
 import { DI } from '@/di-symbols.js';
-import type { UsersRepository, BlockingsRepository, DriveFilesRepository } from '@/models/index.js';
-import type { Config } from '@/config.js';
+import type { UsersRepository, DriveFilesRepository } from '@/models/index.js';
 import type Logger from '@/logger.js';
 import * as Acct from '@/misc/acct.js';
 import { RemoteUserResolveService } from '@/core/RemoteUserResolveService.js';
-import { UserBlockingService } from '@/core/UserBlockingService.js';
 import { DownloadService } from '@/core/DownloadService.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserImportJobData } from '../types.js';
+import type { DbUserImportJobData, DbUserImportToDbJobData } from '../types.js';
 import { bindThis } from '@/decorators.js';
+import { QueueService } from '@/core/QueueService.js';
 
 @Injectable()
 export class ImportBlockingProcessorService {
 	private logger: Logger;
 
 	constructor(
-		@Inject(DI.config)
-		private config: Config,
-
 		@Inject(DI.usersRepository)
 		private usersRepository: UsersRepository,
 
-		@Inject(DI.blockingsRepository)
-		private blockingsRepository: BlockingsRepository,
-
 		@Inject(DI.driveFilesRepository)
 		private driveFilesRepository: DriveFilesRepository,
 
+		private queueService: QueueService,
 		private utilityService: UtilityService,
-		private userBlockingService: UserBlockingService,
 		private remoteUserResolveService: RemoteUserResolveService,
 		private downloadService: DownloadService,
 		private queueLoggerService: QueueLoggerService,
@@ -59,46 +52,50 @@ export class ImportBlockingProcessorService {
 		}
 
 		const csv = await this.downloadService.downloadTextFile(file.url);
+		const targets = csv.trim().split('\n');
+		this.queueService.createImportBlockingToDbJob({ id: user.id }, targets);
 
-		let linenum = 0;
-
-		for (const line of csv.trim().split('\n')) {
-			linenum++;
-
-			try {
-				const acct = line.split(',')[0].trim();
-				const { username, host } = Acct.parse(acct);
-
-				let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
-					host: IsNull(),
-					usernameLower: username.toLowerCase(),
-				}) : await this.usersRepository.findOneBy({
-					host: this.utilityService.toPuny(host!),
-					usernameLower: username.toLowerCase(),
-				});
-
-				if (host == null && target == null) continue;
-
-				if (target == null) {
-					target = await this.remoteUserResolveService.resolveUser(username, host);
-				}
-
-				if (target == null) {
-					throw `cannot resolve user: @${username}@${host}`;
-				}
-
-				// skip myself
-				if (target.id === job.data.user.id) continue;
-
-				this.logger.info(`Block[${linenum}] ${target.id} ...`);
-
-				await this.userBlockingService.block(user, target);
-			} catch (e) {
-				this.logger.warn(`Error in line:${linenum} ${e}`);
-			}
-		}
-
-		this.logger.succ('Imported');
+		this.logger.succ('Import jobs created');
 		done();
 	}
+
+	@bindThis
+	public async processDb(job: Bull.Job<DbUserImportToDbJobData>): Promise<void> {
+		const line = job.data.target;
+		const user = job.data.user;
+
+		try {
+			const acct = line.split(',')[0].trim();
+			const { username, host } = Acct.parse(acct);
+
+			if (!host) return;
+
+			let target = this.utilityService.isSelfHost(host) ? await this.usersRepository.findOneBy({
+				host: IsNull(),
+				usernameLower: username.toLowerCase(),
+			}) : await this.usersRepository.findOneBy({
+				host: this.utilityService.toPuny(host),
+				usernameLower: username.toLowerCase(),
+			});
+
+			if (host == null && target == null) return;
+
+			if (target == null) {
+				target = await this.remoteUserResolveService.resolveUser(username, host);
+			}
+
+			if (target == null) {
+				throw `Unable to resolve user: @${username}@${host}`;
+			}
+
+			// skip myself
+			if (target.id === job.data.user.id) return;
+
+			this.logger.info(`Block ${target.id} ...`);
+
+			this.queueService.createBlockJob([{ from: { id: user.id }, to: { id: target.id }, silent: true }]);
+		} catch (e) {
+			this.logger.warn(`Error: ${e}`);
+		}
+	}
 }

packages/backend/src/queue/processors/ImportFollowingProcessorService.ts

@@ -2,34 +2,30 @@ import { Inject, Injectable } from '@nestjs/common';
 import { IsNull } from 'typeorm';
 import { DI } from '@/di-symbols.js';
 import type { UsersRepository, DriveFilesRepository } from '@/models/index.js';
-import type { Config } from '@/config.js';
 import type Logger from '@/logger.js';
 import * as Acct from '@/misc/acct.js';
 import { RemoteUserResolveService } from '@/core/RemoteUserResolveService.js';
 import { DownloadService } from '@/core/DownloadService.js';
-import { UserFollowingService } from '@/core/UserFollowingService.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
-import type { DbUserImportJobData } from '../types.js';
+import type { DbUserImportJobData, DbUserImportToDbJobData } from '../types.js';
 import { bindThis } from '@/decorators.js';
+import { QueueService } from '@/core/QueueService.js';
 
 @Injectable()
 export class ImportFollowingProcessorService {
 	private logger: Logger;
 
 	constructor(
-		@Inject(DI.config)
-		private config: Config,
-
 		@Inject(DI.usersRepository)
 		private usersRepository: UsersRepository,
 
 		@Inject(DI.driveFilesRepository)
 		private driveFilesRepository: DriveFilesRepository,
 
+		private queueService: QueueService,
 		private utilityService: UtilityService,
-		private userFollowingService: UserFollowingService,
 		private remoteUserResolveService: RemoteUserResolveService,
 		private downloadService: DownloadService,
 		private queueLoggerService: QueueLoggerService,
@@ -56,46 +52,50 @@ export class ImportFollowingProcessorService {
 		}
 
 		const csv = await this.downloadService.downloadTextFile(file.url);
+		const targets = csv.trim().split('\n');
+		this.queueService.createImportFollowingToDbJob({ id: user.id }, targets);
 
-		let linenum = 0;
-
-		for (const line of csv.trim().split('\n')) {
-			linenum++;
-
-			try {
-				const acct = line.split(',')[0].trim();
-				const { username, host } = Acct.parse(acct);
-
-				let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
-					host: IsNull(),
-					usernameLower: username.toLowerCase(),
-				}) : await this.usersRepository.findOneBy({
-					host: this.utilityService.toPuny(host!),
-					usernameLower: username.toLowerCase(),
-				});
-
-				if (host == null && target == null) continue;
-
-				if (target == null) {
-					target = await this.remoteUserResolveService.resolveUser(username, host);
-				}
-
-				if (target == null) {
-					throw `cannot resolve user: @${username}@${host}`;
-				}
-
-				// skip myself
-				if (target.id === job.data.user.id) continue;
-
-				this.logger.info(`Follow[${linenum}] ${target.id} ...`);
-
-				this.userFollowingService.follow(user, target);
-			} catch (e) {
-				this.logger.warn(`Error in line:${linenum} ${e}`);
-			}
-		}
-
-		this.logger.succ('Imported');
+		this.logger.succ('Import jobs created');
 		done();
 	}
+
+	@bindThis
+	public async processDb(job: Bull.Job<DbUserImportToDbJobData>): Promise<void> {
+		const line = job.data.target;
+		const user = job.data.user;
+
+		try {
+			const acct = line.split(',')[0].trim();
+			const { username, host } = Acct.parse(acct);
+
+			if (!host) return;
+
+			let target = this.utilityService.isSelfHost(host) ? await this.usersRepository.findOneBy({
+				host: IsNull(),
+				usernameLower: username.toLowerCase(),
+			}) : await this.usersRepository.findOneBy({
+				host: this.utilityService.toPuny(host),
+				usernameLower: username.toLowerCase(),
+			});
+
+			if (host == null && target == null) return;
+
+			if (target == null) {
+				target = await this.remoteUserResolveService.resolveUser(username, host);
+			}
+
+			if (target == null) {
+				throw `Unable to resolve user: @${username}@${host}`;
+			}
+
+			// skip myself
+			if (target.id === job.data.user.id) return;
+
+			this.logger.info(`Follow ${target.id} ...`);
+
+			this.queueService.createFollowJob([{ from: user, to: { id: target.id }, silent: true }]);
+		} catch (e) {
+			this.logger.warn(`Error: ${e}`);
+		}
+	}
 }

packages/backend/src/queue/processors/ImportMutingProcessorService.ts

@@ -66,11 +66,13 @@ export class ImportMutingProcessorService {
 				const acct = line.split(',')[0].trim();
 				const { username, host } = Acct.parse(acct);
 
-				let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
+				if (!host) continue;
+
+				let target = this.utilityService.isSelfHost(host) ? await this.usersRepository.findOneBy({
 					host: IsNull(),
 					usernameLower: username.toLowerCase(),
 				}) : await this.usersRepository.findOneBy({
-					host: this.utilityService.toPuny(host!),
+					host: this.utilityService.toPuny(host),
 					usernameLower: username.toLowerCase(),
 				});
 

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';
+	}
+}

packages/backend/src/queue/types.ts

@@ -21,9 +21,39 @@ export type InboxJobData = {
 	signature: httpSignature.IParsedSignature;
 };
 
-export type DbJobData = DbUserJobData | DbUserImportJobData | DbUserDeleteJobData;
+export type RelationshipJobData = {
+	from: ThinUser;
+	to: ThinUser;
+	silent?: boolean;
+	requestId?: string;
+}
 
-export type DbUserJobData = {
+export type DbJobData<T extends keyof DbJobMap> = DbJobMap[T];
+
+export type DbJobMap = {
+	deleteDriveFiles: DbJobDataWithUser;
+	exportCustomEmojis: DbJobDataWithUser;
+	exportNotes: DbJobDataWithUser;
+	exportFavorites: DbJobDataWithUser;
+	exportFollowing: DbExportFollowingData;
+	exportMuting: DbJobDataWithUser;
+	exportBlocking: DbJobDataWithUser;
+	exportUserLists: DbJobDataWithUser;
+	importFollowing: DbUserImportJobData;
+	importFollowingToDb: DbUserImportToDbJobData;
+	importMuting: DbUserImportJobData;
+	importBlocking: DbUserImportJobData;
+	importBlockingToDb: DbUserImportToDbJobData;
+	importUserLists: DbUserImportJobData;
+	importCustomEmojis: DbUserImportJobData;
+	deleteAccount: DbUserDeleteJobData;
+}
+
+export type DbJobDataWithUser = {
+	user: ThinUser;
+}
+
+export type DbExportFollowingData = {
 	user: ThinUser;
 	excludeMuting: boolean;
 	excludeInactive: boolean;
@@ -39,6 +69,11 @@ export type DbUserImportJobData = {
 	fileId: DriveFile['id'];
 };
 
+export type DbUserImportToDbJobData = {
+	user: ThinUser;
+	target: string;
+};
+
 export type ObjectStorageJobData = ObjectStorageFileJobData | Record<string, unknown>;
 
 export type ObjectStorageFileJobData = {