Resolve #4444

9 files changed, 75 insertions, 87 deletions

diff --git a/src/queue/processors/delete-drive-files.ts b/src/queue/processors/db/delete-drive-files.ts
index 7e76aa73e6..3de960a25e 100644
--- a/src/queue/processors/delete-drive-files.ts
+++ b/src/queue/processors/db/delete-drive-files.ts
@@ -1,14 +1,14 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as mongo from 'mongodb';
 
-import { queueLogger } from '../logger';
-import User from '../../models/user';
-import DriveFile from '../../models/drive-file';
-import deleteFile from '../../services/drive/delete-file';
+import { queueLogger } from '../../logger';
+import User from '../../../models/user';
+import DriveFile from '../../../models/drive-file';
+import deleteFile from '../../../services/drive/delete-file';
 
 const logger = queueLogger.createSubLogger('delete-drive-files');
 
-export async function deleteDriveFiles(job: bq.Job, done: any): Promise<void> {
+export async function deleteDriveFiles(job: Bull.Job, done: any): Promise<void> {
 	logger.info(`Deleting drive files of ${job.data.user._id} ...`);
 
 	const user = await User.findOne({
@@ -32,7 +32,7 @@ export async function deleteDriveFiles(job: bq.Job, done: any): Promise<void> {
 
 		if (files.length === 0) {
 			ended = true;
-			if (job.reportProgress) job.reportProgress(100);
+			job.progress(100);
 			break;
 		}
 
@@ -47,7 +47,7 @@ export async function deleteDriveFiles(job: bq.Job, done: any): Promise<void> {
 			userId: user._id,
 		});
 
-		if (job.reportProgress) job.reportProgress(deletedCount / total);
+		job.progress(deletedCount / total);
 	}
 
 	logger.succ(`All drive files (${deletedCount}) of ${user._id} has been deleted.`);
diff --git a/src/queue/processors/delete-notes.ts b/src/queue/processors/db/delete-notes.ts
index 13c6042b16..021db8062e 100644
--- a/src/queue/processors/delete-notes.ts
+++ b/src/queue/processors/db/delete-notes.ts
@@ -1,14 +1,14 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as mongo from 'mongodb';
 
-import { queueLogger } from '../logger';
-import Note from '../../models/note';
-import deleteNote from '../../services/note/delete';
-import User from '../../models/user';
+import { queueLogger } from '../../logger';
+import Note from '../../../models/note';
+import deleteNote from '../../../services/note/delete';
+import User from '../../../models/user';
 
 const logger = queueLogger.createSubLogger('delete-notes');
 
-export async function deleteNotes(job: bq.Job, done: any): Promise<void> {
+export async function deleteNotes(job: Bull.Job, done: any): Promise<void> {
 	logger.info(`Deleting notes of ${job.data.user._id} ...`);
 
 	const user = await User.findOne({
@@ -32,7 +32,7 @@ export async function deleteNotes(job: bq.Job, done: any): Promise<void> {
 
 		if (notes.length === 0) {
 			ended = true;
-			if (job.reportProgress) job.reportProgress(100);
+			job.progress(100);
 			break;
 		}
 
@@ -47,7 +47,7 @@ export async function deleteNotes(job: bq.Job, done: any): Promise<void> {
 			userId: user._id,
 		});
 
-		if (job.reportProgress) job.reportProgress(deletedCount / total);
+		job.progress(deletedCount / total);
 	}
 
 	logger.succ(`All notes (${deletedCount}) of ${user._id} has been deleted.`);
diff --git a/src/queue/processors/export-blocking.ts b/src/queue/processors/db/export-blocking.ts
index b30d8e3bc8..e56aec94ac 100644
--- a/src/queue/processors/export-blocking.ts
+++ b/src/queue/processors/db/export-blocking.ts
@@ -1,18 +1,18 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as tmp from 'tmp';
 import * as fs from 'fs';
 import * as mongo from 'mongodb';
 
-import { queueLogger } from '../logger';
-import addFile from '../../services/drive/add-file';
-import User from '../../models/user';
+import { queueLogger } from '../../logger';
+import addFile from '../../../services/drive/add-file';
+import User from '../../../models/user';
 import dateFormat = require('dateformat');
-import Blocking from '../../models/blocking';
-import config from '../../config';
+import Blocking from '../../../models/blocking';
+import config from '../../../config';
 
 const logger = queueLogger.createSubLogger('export-blocking');
 
-export async function exportBlocking(job: bq.Job, done: any): Promise<void> {
+export async function exportBlocking(job: Bull.Job, done: any): Promise<void> {
 	logger.info(`Exporting blocking of ${job.data.user._id} ...`);
 
 	const user = await User.findOne({
@@ -48,7 +48,7 @@ export async function exportBlocking(job: bq.Job, done: any): Promise<void> {
 
 		if (blockings.length === 0) {
 			ended = true;
-			if (job.reportProgress) job.reportProgress(100);
+			job.progress(100);
 			break;
 		}
 
@@ -74,7 +74,7 @@ export async function exportBlocking(job: bq.Job, done: any): Promise<void> {
 			blockerId: user._id,
 		});
 
-		if (job.reportProgress) job.reportProgress(exportedCount / total);
+		job.progress(exportedCount / total);
 	}
 
 	stream.end();
diff --git a/src/queue/processors/export-following.ts b/src/queue/processors/db/export-following.ts
index e6521d0652..1d8a501b78 100644
--- a/src/queue/processors/export-following.ts
+++ b/src/queue/processors/db/export-following.ts
@@ -1,18 +1,18 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as tmp from 'tmp';
 import * as fs from 'fs';
 import * as mongo from 'mongodb';
 
-import { queueLogger } from '../logger';
-import addFile from '../../services/drive/add-file';
-import User from '../../models/user';
+import { queueLogger } from '../../logger';
+import addFile from '../../../services/drive/add-file';
+import User from '../../../models/user';
 import dateFormat = require('dateformat');
-import Following from '../../models/following';
-import config from '../../config';
+import Following from '../../../models/following';
+import config from '../../../config';
 
 const logger = queueLogger.createSubLogger('export-following');
 
-export async function exportFollowing(job: bq.Job, done: any): Promise<void> {
+export async function exportFollowing(job: Bull.Job, done: any): Promise<void> {
 	logger.info(`Exporting following of ${job.data.user._id} ...`);
 
 	const user = await User.findOne({
@@ -48,7 +48,7 @@ export async function exportFollowing(job: bq.Job, done: any): Promise<void> {
 
 		if (followings.length === 0) {
 			ended = true;
-			if (job.reportProgress) job.reportProgress(100);
+			job.progress(100);
 			break;
 		}
 
@@ -74,7 +74,7 @@ export async function exportFollowing(job: bq.Job, done: any): Promise<void> {
 			followerId: user._id,
 		});
 
-		if (job.reportProgress) job.reportProgress(exportedCount / total);
+		job.progress(exportedCount / total);
 	}
 
 	stream.end();
diff --git a/src/queue/processors/export-mute.ts b/src/queue/processors/db/export-mute.ts
index 74456c1da2..6f2dd6df13 100644
--- a/src/queue/processors/export-mute.ts
+++ b/src/queue/processors/db/export-mute.ts
@@ -1,18 +1,18 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as tmp from 'tmp';
 import * as fs from 'fs';
 import * as mongo from 'mongodb';
 
-import { queueLogger } from '../logger';
-import addFile from '../../services/drive/add-file';
-import User from '../../models/user';
+import { queueLogger } from '../../logger';
+import addFile from '../../../services/drive/add-file';
+import User from '../../../models/user';
 import dateFormat = require('dateformat');
-import Mute from '../../models/mute';
-import config from '../../config';
+import Mute from '../../../models/mute';
+import config from '../../../config';
 
 const logger = queueLogger.createSubLogger('export-mute');
 
-export async function exportMute(job: bq.Job, done: any): Promise<void> {
+export async function exportMute(job: Bull.Job, done: any): Promise<void> {
 	logger.info(`Exporting mute of ${job.data.user._id} ...`);
 
 	const user = await User.findOne({
@@ -48,7 +48,7 @@ export async function exportMute(job: bq.Job, done: any): Promise<void> {
 
 		if (mutes.length === 0) {
 			ended = true;
-			if (job.reportProgress) job.reportProgress(100);
+			job.progress(100);
 			break;
 		}
 
@@ -74,7 +74,7 @@ export async function exportMute(job: bq.Job, done: any): Promise<void> {
 			muterId: user._id,
 		});
 
-		if (job.reportProgress) job.reportProgress(exportedCount / total);
+		job.progress(exportedCount / total);
 	}
 
 	stream.end();
diff --git a/src/queue/processors/export-notes.ts b/src/queue/processors/db/export-notes.ts
index 32e4cd1d6c..8f3cdc5b99 100644
--- a/src/queue/processors/export-notes.ts
+++ b/src/queue/processors/db/export-notes.ts
@@ -1,17 +1,17 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as tmp from 'tmp';
 import * as fs from 'fs';
 import * as mongo from 'mongodb';
 
-import { queueLogger } from '../logger';
-import Note, { INote } from '../../models/note';
-import addFile from '../../services/drive/add-file';
-import User from '../../models/user';
+import { queueLogger } from '../../logger';
+import Note, { INote } from '../../../models/note';
+import addFile from '../../../services/drive/add-file';
+import User from '../../../models/user';
 import dateFormat = require('dateformat');
 
 const logger = queueLogger.createSubLogger('export-notes');
 
-export async function exportNotes(job: bq.Job, done: any): Promise<void> {
+export async function exportNotes(job: Bull.Job, done: any): Promise<void> {
 	logger.info(`Exporting notes of ${job.data.user._id} ...`);
 
 	const user = await User.findOne({
@@ -58,7 +58,7 @@ export async function exportNotes(job: bq.Job, done: any): Promise<void> {
 
 		if (notes.length === 0) {
 			ended = true;
-			if (job.reportProgress) job.reportProgress(100);
+			job.progress(100);
 			break;
 		}
 
@@ -83,7 +83,7 @@ export async function exportNotes(job: bq.Job, done: any): Promise<void> {
 			userId: user._id,
 		});
 
-		if (job.reportProgress) job.reportProgress(exportedNotesCount / total);
+		job.progress(exportedNotesCount / total);
 	}
 
 	await new Promise((res, rej) => {
diff --git a/src/queue/processors/index.ts b/src/queue/processors/db/index.ts
index 31e87c3f67..91d7f06a4b 100644
--- a/src/queue/processors/index.ts
+++ b/src/queue/processors/db/index.ts
@@ -1,31 +1,20 @@
-import deliver from './http/deliver';
-import processInbox from './http/process-inbox';
+import * as Bull from 'bull';
 import { deleteNotes } from './delete-notes';
 import { deleteDriveFiles } from './delete-drive-files';
 import { exportNotes } from './export-notes';
 import { exportFollowing } from './export-following';
 import { exportMute } from './export-mute';
 import { exportBlocking } from './export-blocking';
-import { queueLogger } from '../logger';
 
-const handlers: any = {
-	deliver,
-	processInbox,
+const jobs = {
 	deleteNotes,
 	deleteDriveFiles,
 	exportNotes,
 	exportFollowing,
 	exportMute,
 	exportBlocking,
-};
+} as any;
 
-export default (job: any, done: any) => {
-	const handler = handlers[job.data.type];
-
-	if (handler) {
-		handler(job, done);
-	} else {
-		queueLogger.error(`Unknown job: ${job.data.type}`);
-		done();
-	}
-};
+export default function(job: Bull.Job, done: any) {
+	jobs[job.data.type](job, done);
+}
diff --git a/src/queue/processors/http/deliver.ts b/src/queue/processors/deliver.ts
index 96f6cc07ce..2f730c31cd 100644
--- a/src/queue/processors/http/deliver.ts
+++ b/src/queue/processors/deliver.ts
@@ -1,14 +1,13 @@
-import * as bq from 'bee-queue';
-
-import request from '../../../remote/activitypub/request';
-import { queueLogger } from '../../logger';
-import { registerOrFetchInstanceDoc } from '../../../services/register-or-fetch-instance-doc';
-import Instance from '../../../models/instance';
-import instanceChart from '../../../services/chart/instance';
+import * as Bull from 'bull';
+import request from '../../remote/activitypub/request';
+import { queueLogger } from '../logger';
+import { registerOrFetchInstanceDoc } from '../../services/register-or-fetch-instance-doc';
+import Instance from '../../models/instance';
+import instanceChart from '../../services/chart/instance';
 
 let latest: string = null;
 
-export default async (job: bq.Job, done: any): Promise<void> => {
+export default async (job: Bull.Job, done: any): Promise<void> => {
 	const { host } = new URL(job.data.to);
 
 	try {
diff --git a/src/queue/processors/http/process-inbox.ts b/src/queue/processors/process-inbox.ts
index cc4e711d00..21292921f0 100644
--- a/src/queue/processors/http/process-inbox.ts
+++ b/src/queue/processors/process-inbox.ts
@@ -1,21 +1,21 @@
-import * as bq from 'bee-queue';
+import * as Bull from 'bull';
 import * as httpSignature from 'http-signature';
-import parseAcct from '../../../misc/acct/parse';
-import User, { IRemoteUser } from '../../../models/user';
-import perform from '../../../remote/activitypub/perform';
-import { resolvePerson, updatePerson } from '../../../remote/activitypub/models/person';
+import parseAcct from '../../misc/acct/parse';
+import User, { IRemoteUser } from '../../models/user';
+import perform from '../../remote/activitypub/perform';
+import { resolvePerson, updatePerson } from '../../remote/activitypub/models/person';
 import { toUnicode } from 'punycode';
 import { URL } from 'url';
-import { publishApLogStream } from '../../../services/stream';
-import Logger from '../../../services/logger';
-import { registerOrFetchInstanceDoc } from '../../../services/register-or-fetch-instance-doc';
-import Instance from '../../../models/instance';
-import instanceChart from '../../../services/chart/instance';
+import { publishApLogStream } from '../../services/stream';
+import Logger from '../../services/logger';
+import { registerOrFetchInstanceDoc } from '../../services/register-or-fetch-instance-doc';
+import Instance from '../../models/instance';
+import instanceChart from '../../services/chart/instance';
 
 const logger = new Logger('inbox');
 
 // ユーザーのinboxにアクティビティが届いた時の処理
-export default async (job: bq.Job, done: any): Promise<void> => {
+export default async (job: Bull.Job, done: any): Promise<void> => {
 	const signature = job.data.signature;
 	const activity = job.data.activity;