Resolve #4444

12 files changed, 144 insertions, 173 deletions

diff --git a/src/argv.ts b/src/argv.ts
index d9cd12dfcc..b5540441cc 100644
--- a/src/argv.ts
+++ b/src/argv.ts
@@ -5,8 +5,7 @@ program
 	.version(pkg.version)
 	.option('--no-daemons', 'Disable daemon processes (for debbuging)')
 	.option('--disable-clustering', 'Disable clustering')
-	.option('--disable-queue', 'Disable job queue processing')
-	.option('--only-server', 'Run server only (without job queue)')
+	.option('--only-server', 'Run server only (without job queue processing)')
 	.option('--only-queue', 'Pocessing job queue only (without server)')
 	.option('--quiet', 'Suppress all logs')
 	.option('--verbose', 'Enable all logs')
@@ -15,7 +14,6 @@ program
 	.option('--color', 'This option is a dummy for some external program\'s (e.g. forever) issue.')
 	.parse(process.argv);
 
-/*if (process.env.MK_DISABLE_QUEUE)*/ program.disableQueue = true;
 if (process.env.MK_ONLY_QUEUE) program.onlyQueue = true;
 
 export { program };
diff --git a/src/queue/index.ts b/src/queue/index.ts
index 351a035ada..bb3b66908d 100644
--- a/src/queue/index.ts
+++ b/src/queue/index.ts
@@ -1,73 +1,64 @@
-import * as Queue from 'bee-queue';
+import * as Queue from 'bull';
 import * as httpSignature from 'http-signature';
 
 import config from '../config';
 import { ILocalUser } from '../models/user';
 import { program } from '../argv';
-import handler from './processors';
-import { queueLogger } from './logger';
 
-const enableQueue = !program.disableQueue;
-const enableQueueProcessing = !program.onlyServer && enableQueue;
-const queueAvailable = config.redis != null;
+import processDeliver from './processors/deliver';
+import processInbox from './processors/process-inbox';
+import processDb from './processors/db';
 
-const queue = initializeQueue();
-
-function initializeQueue() {
-	if (queueAvailable && enableQueue) {
-		return new Queue('misskey-queue', {
-			redis: {
-				port: config.redis.port,
-				host: config.redis.host,
-				password: config.redis.pass
-			},
-
-			removeOnSuccess: true,
-			removeOnFailure: true,
-			getEvents: false,
-			sendEvents: false,
-			storeJobs: false
-		});
-	} else {
-		return null;
-	}
+function initializeQueue(name: string) {
+	return new Queue(name, config.redis != null ? {
+		redis: {
+			port: config.redis.port,
+			host: config.redis.host,
+			password: config.redis.pass,
+			db: 1
+		}
+	} : null);
 }
 
+const deliverQueue = initializeQueue('deliver');
+const inboxQueue = initializeQueue('inbox');
+const dbQueue = initializeQueue('db');
+
 export function deliver(user: ILocalUser, content: any, to: any) {
-	if (content == null) return;
+	if (content == null) return null;
 
 	const data = {
-		type: 'deliver',
 		user,
 		content,
 		to
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data)
-			.retries(8)
-			.backoff('exponential', 1000)
-			.save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return deliverQueue.add(data, {
+		attempts: 4,
+		backoff: {
+			type: 'exponential',
+			delay: 1000
+		},
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
-export function processInbox(activity: any, signature: httpSignature.IParsedSignature) {
+export function inbox(activity: any, signature: httpSignature.IParsedSignature) {
 	const data = {
-		type: 'processInbox',
 		activity: activity,
 		signature
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data)
-			.retries(3)
-			.backoff('exponential', 500)
-			.save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return inboxQueue.add(data, {
+		attempts: 4,
+		backoff: {
+			type: 'exponential',
+			delay: 1000
+		},
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export function createDeleteNotesJob(user: ILocalUser) {
@@ -76,11 +67,10 @@ export function createDeleteNotesJob(user: ILocalUser) {
 		user: user
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data).save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return dbQueue.add(data, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export function createDeleteDriveFilesJob(user: ILocalUser) {
@@ -89,11 +79,10 @@ export function createDeleteDriveFilesJob(user: ILocalUser) {
 		user: user
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data).save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return dbQueue.add(data, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export function createExportNotesJob(user: ILocalUser) {
@@ -102,11 +91,10 @@ export function createExportNotesJob(user: ILocalUser) {
 		user: user
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data).save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return dbQueue.add(data, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export function createExportFollowingJob(user: ILocalUser) {
@@ -115,11 +103,10 @@ export function createExportFollowingJob(user: ILocalUser) {
 		user: user
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data).save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return dbQueue.add(data, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export function createExportMuteJob(user: ILocalUser) {
@@ -128,11 +115,10 @@ export function createExportMuteJob(user: ILocalUser) {
 		user: user
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data).save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return dbQueue.add(data, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export function createExportBlockingJob(user: ILocalUser) {
@@ -141,24 +127,23 @@ export function createExportBlockingJob(user: ILocalUser) {
 		user: user
 	};
 
-	if (queueAvailable && enableQueueProcessing) {
-		return queue.createJob(data).save();
-	} else {
-		return handler({ data }, () => {});
-	}
+	return dbQueue.add(data, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
 }
 
 export default function() {
-	if (queueAvailable && enableQueueProcessing) {
-		queue.process(128, handler);
-		queueLogger.succ('Processing started');
+	if (!program.onlyServer) {
+		deliverQueue.process(processDeliver);
+		inboxQueue.process(processInbox);
+		dbQueue.process(processDb);
 	}
-
-	return queue;
 }
 
 export function destroy() {
+	/*
 	queue.destroy().then(n => {
 		queueLogger.succ(`All job removed (${n} jobs)`);
-	});
+	});*/
 }
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;
 
diff --git a/src/server/activitypub.ts b/src/server/activitypub.ts
index 24c8e556b6..2151ada344 100644
--- a/src/server/activitypub.ts
+++ b/src/server/activitypub.ts
@@ -16,7 +16,7 @@ import Followers from './activitypub/followers';
 import Following from './activitypub/following';
 import Featured from './activitypub/featured';
 import renderQuestion from '../remote/activitypub/renderer/question';
-import { processInbox } from '../queue';
+import { inbox as processInbox } from '../queue';
 
 // Init router
 const router = new Router();