Resolve #4444

package.json

@@ -32,6 +32,7 @@
 		"@prezzemolo/rap": "0.1.2",
 		"@prezzemolo/zip": "0.0.3",
 		"@types/bcryptjs": "2.4.2",
+		"@types/bull": "3.5.8",
 		"@types/chai-http": "3.0.5",
 		"@types/dateformat": "3.0.0",
 		"@types/deep-equal": "1.0.1",
@@ -100,8 +101,8 @@
 		"autosize": "4.0.2",
 		"autwh": "0.1.0",
 		"bcryptjs": "2.4.3",
-		"bee-queue": "1.2.2",
 		"bootstrap-vue": "2.0.0-rc.13",
+		"bull": "3.7.0",
 		"cafy": "15.1.0",
 		"chai": "4.2.0",
 		"chai-http": "4.2.1",

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

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', {
+function initializeQueue(name: string) {
+	return new Queue(name, config.redis != null ? {
 		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;
+			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)`);
-	});
+	});*/
 }

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.`);

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.`);

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();

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();

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();

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) => {

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

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 {

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;
 

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();