enhance(backend): migrate bull to bullmq (#10910)

* wip * wip * Update QueueService.ts * wip * refactor * ✌️ * fix * Update QueueStatsService.ts * refactor * Update ApNoteService.ts * Update mock-resolver.ts * refactor * Update mock-resolver.ts
2023-05-29 11:54:49 +09:00
parent 7cbd852fe5
commit fd7b77c542
53 changed files with 532 additions and 490 deletions
--- a/packages/backend/src/core/CaptchaService.ts
+++ b/packages/backend/src/core/CaptchaService.ts
@@ -30,7 +30,7 @@ export class CaptchaService {
 		}, { throwErrorWhenResponseNotOk: false });
 	
 		if (!res.ok) {
-			throw `${res.status}`;
+			throw new Error(`${res.status}`);
 		}
 	
 		return await res.json() as CaptchaResponse;
@@ -39,48 +39,48 @@ export class CaptchaService {
 	@bindThis
 	public async verifyRecaptcha(secret: string, response: string | null | undefined): Promise<void> {
 		if (response == null) {
-			throw 'recaptcha-failed: no response provided';
+			throw new Error('recaptcha-failed: no response provided');
 		}

 		const result = await this.getCaptchaResponse('https://www.recaptcha.net/recaptcha/api/siteverify', secret, response).catch(err => {
-			throw `recaptcha-request-failed: ${err}`;
+			throw new Error(`recaptcha-request-failed: ${err}`);
 		});

 		if (result.success !== true) {
 			const errorCodes = result['error-codes'] ? result['error-codes'].join(', ') : '';
-			throw `recaptcha-failed: ${errorCodes}`;
+			throw new Error(`recaptcha-failed: ${errorCodes}`);
 		}
 	}

 	@bindThis
 	public async verifyHcaptcha(secret: string, response: string | null | undefined): Promise<void> {
 		if (response == null) {
-			throw 'hcaptcha-failed: no response provided';
+			throw new Error('hcaptcha-failed: no response provided');
 		}

 		const result = await this.getCaptchaResponse('https://hcaptcha.com/siteverify', secret, response).catch(err => {
-			throw `hcaptcha-request-failed: ${err}`;
+			throw new Error(`hcaptcha-request-failed: ${err}`);
 		});

 		if (result.success !== true) {
 			const errorCodes = result['error-codes'] ? result['error-codes'].join(', ') : '';
-			throw `hcaptcha-failed: ${errorCodes}`;
+			throw new Error(`hcaptcha-failed: ${errorCodes}`);
 		}
 	}

 	@bindThis
 	public async verifyTurnstile(secret: string, response: string | null | undefined): Promise<void> {
 		if (response == null) {
-			throw 'turnstile-failed: no response provided';
+			throw new Error('turnstile-failed: no response provided');
 		}
 	
 		const result = await this.getCaptchaResponse('https://challenges.cloudflare.com/turnstile/v0/siteverify', secret, response).catch(err => {
-			throw `turnstile-request-failed: ${err}`;
+			throw new Error(`turnstile-request-failed: ${err}`);
 		});

 		if (result.success !== true) {
 			const errorCodes = result['error-codes'] ? result['error-codes'].join(', ') : '';
-			throw `turnstile-failed: ${errorCodes}`;
+			throw new Error(`turnstile-failed: ${errorCodes}`);
 		}
 	}
 }
--- a/packages/backend/src/core/FetchInstanceMetadataService.ts
+++ b/packages/backend/src/core/FetchInstanceMetadataService.ts
@@ -116,14 +116,14 @@ export class FetchInstanceMetadataService {
 			const wellknown = await this.httpRequestService.getJson('https://' + instance.host + '/.well-known/nodeinfo')
 				.catch(err => {
 					if (err.statusCode === 404) {
-						throw 'No nodeinfo provided';
+						throw new Error('No nodeinfo provided');
 					} else {
 						throw err.statusCode ?? err.message;
 					}
 				}) as Record<string, unknown>;
 	
 			if (wellknown.links == null || !Array.isArray(wellknown.links)) {
-				throw 'No wellknown links';
+				throw new Error('No wellknown links');
 			}
 	
 			const links = wellknown.links as any[];
@@ -134,7 +134,7 @@ export class FetchInstanceMetadataService {
 			const link = lnik2_1 ?? lnik2_0 ?? lnik1_0;
 	
 			if (link == null) {
-				throw 'No nodeinfo link provided';
+				throw new Error('No nodeinfo link provided');
 			}
 	
 			const info = await this.httpRequestService.getJson(link.href)
--- a/packages/backend/src/core/NoteCreateService.ts
+++ b/packages/backend/src/core/NoteCreateService.ts
@@ -510,7 +510,7 @@ export class NoteCreateService implements OnApplicationShutdown {

 		if (data.poll && data.poll.expiresAt) {
 			const delay = data.poll.expiresAt.getTime() - Date.now();
-			this.queueService.endedPollNotificationQueue.add({
+			this.queueService.endedPollNotificationQueue.add(note.id, {
 				noteId: note.id,
 			}, {
 				delay,
--- a/packages/backend/src/core/QueueModule.ts
+++ b/packages/backend/src/core/QueueModule.ts
@@ -1,42 +1,11 @@
 import { setTimeout } from 'node:timers/promises';
 import { Inject, Module, OnApplicationShutdown } from '@nestjs/common';
-import Bull from 'bull';
+import * as Bull from 'bullmq';
 import { DI } from '@/di-symbols.js';
 import type { Config } from '@/config.js';
+import { QUEUE, baseQueueOptions } from '@/queue/const.js';
 import type { Provider } from '@nestjs/common';
-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, {
-		redis: {
-			port: config.redisForJobQueue.port,
-			host: config.redisForJobQueue.host,
-			family: config.redisForJobQueue.family == null ? 0 : config.redisForJobQueue.family,
-			password: config.redisForJobQueue.pass,
-			db: config.redisForJobQueue.db ?? 0,
-		},
-		prefix: config.redisForJobQueue.prefix ? `${config.redisForJobQueue.prefix}:queue` : 'queue',
-		limiter: limitPerSec > 0 ? {
-			max: limitPerSec,
-			duration: 1000,
-		} : undefined,
-		settings: {
-			backoffStrategies: {
-				apBackoff,
-			},
-		},
-	});
-}
-
-// ref. https://github.com/misskey-dev/misskey/pull/7635#issue-971097019
-function apBackoff(attemptsMade: number, err: Error) {
-	const baseDelay = 60 * 1000;	// 1min
-	const maxBackoff = 8 * 60 * 60 * 1000;	// 8hours
-	let backoff = (Math.pow(2, attemptsMade) - 1) * baseDelay;
-	backoff = Math.min(backoff, maxBackoff);
-	backoff += Math.round(backoff * Math.random() * 0.2);
-	return backoff;
-}
+import type { DeliverJobData, InboxJobData, EndedPollNotificationJobData, WebhookDeliverJobData, RelationshipJobData } from '../queue/types.js';

 export type SystemQueue = Bull.Queue<Record<string, unknown>>;
 export type EndedPollNotificationQueue = Bull.Queue<EndedPollNotificationJobData>;
@@ -49,49 +18,49 @@ export type WebhookDeliverQueue = Bull.Queue<WebhookDeliverJobData>;

 const $system: Provider = {
 	provide: 'queue:system',
-	useFactory: (config: Config) => q(config, 'system'),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM)),
 	inject: [DI.config],
 };

 const $endedPollNotification: Provider = {
 	provide: 'queue:endedPollNotification',
-	useFactory: (config: Config) => q(config, 'endedPollNotification'),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.ENDED_POLL_NOTIFICATION, baseQueueOptions(config, QUEUE.ENDED_POLL_NOTIFICATION)),
 	inject: [DI.config],
 };

 const $deliver: Provider = {
 	provide: 'queue:deliver',
-	useFactory: (config: Config) => q(config, 'deliver', config.deliverJobPerSec ?? 128),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER)),
 	inject: [DI.config],
 };

 const $inbox: Provider = {
 	provide: 'queue:inbox',
-	useFactory: (config: Config) => q(config, 'inbox', config.inboxJobPerSec ?? 16),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX)),
 	inject: [DI.config],
 };

 const $db: Provider = {
 	provide: 'queue:db',
-	useFactory: (config: Config) => q(config, 'db'),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB)),
 	inject: [DI.config],
 };

 const $relationship: Provider = {
 	provide: 'queue:relationship',
-	useFactory: (config: Config) => q(config, 'relationship', config.relashionshipJobPerSec ?? 64),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP)),
 	inject: [DI.config],
 };

 const $objectStorage: Provider = {
 	provide: 'queue:objectStorage',
-	useFactory: (config: Config) => q(config, 'objectStorage'),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.OBJECT_STORAGE, baseQueueOptions(config, QUEUE.OBJECT_STORAGE)),
 	inject: [DI.config],
 };

 const $webhookDeliver: Provider = {
 	provide: 'queue:webhookDeliver',
-	useFactory: (config: Config) => q(config, 'webhookDeliver', 64),
+	useFactory: (config: Config) => new Bull.Queue(QUEUE.WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.WEBHOOK_DELIVER)),
 	inject: [DI.config],
 };

--- a/packages/backend/src/core/QueueService.ts
+++ b/packages/backend/src/core/QueueService.ts
@@ -1,6 +1,5 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { v4 as uuid } from 'uuid';
-import Bull from 'bull';
 import type { IActivity } from '@/core/activitypub/type.js';
 import type { DriveFile } from '@/models/entities/DriveFile.js';
 import type { Webhook, webhookEventTypes } from '@/models/entities/Webhook.js';
@@ -11,6 +10,7 @@ import type { Antenna } from '@/server/api/endpoints/i/import-antennas.js';
 import type { DbQueue, DeliverQueue, EndedPollNotificationQueue, InboxQueue, ObjectStorageQueue, RelationshipQueue, SystemQueue, WebhookDeliverQueue } from './QueueModule.js';
 import type { DbJobData, RelationshipJobData, ThinUser } from '../queue/types.js';
 import type httpSignature from '@peertube/http-signature';
+import type * as Bull from 'bullmq';

@Injectable()
 export class QueueService {
@@ -26,7 +26,43 @@ export class QueueService {
 		@Inject('queue:relationship') public relationshipQueue: RelationshipQueue,
 		@Inject('queue:objectStorage') public objectStorageQueue: ObjectStorageQueue,
 		@Inject('queue:webhookDeliver') public webhookDeliverQueue: WebhookDeliverQueue,
-	) {}
+	) {
+		this.systemQueue.add('tickCharts', {
+		}, {
+			repeat: { pattern: '55 * * * *' },
+			removeOnComplete: true,
+		});
+
+		this.systemQueue.add('resyncCharts', {
+		}, {
+			repeat: { pattern: '0 0 * * *' },
+			removeOnComplete: true,
+		});
+
+		this.systemQueue.add('cleanCharts', {
+		}, {
+			repeat: { pattern: '0 0 * * *' },
+			removeOnComplete: true,
+		});
+
+		this.systemQueue.add('aggregateRetention', {
+		}, {
+			repeat: { pattern: '0 0 * * *' },
+			removeOnComplete: true,
+		});
+
+		this.systemQueue.add('clean', {
+		}, {
+			repeat: { pattern: '0 0 * * *' },
+			removeOnComplete: true,
+		});
+
+		this.systemQueue.add('checkExpiredMutings', {
+		}, {
+			repeat: { pattern: '*/5 * * * *' },
+			removeOnComplete: true,
+		});
+	}

 	@bindThis
 	public deliver(user: ThinUser, content: IActivity | null, to: string | null, isSharedInbox: boolean) {
@@ -42,11 +78,10 @@ export class QueueService {
 			isSharedInbox,
 		};

-		return this.deliverQueue.add(data, {
+		return this.deliverQueue.add(to, data, {
 			attempts: this.config.deliverJobMaxAttempts ?? 12,
-			timeout: 1 * 60 * 1000,	// 1min
 			backoff: {
-				type: 'apBackoff',
+				type: 'custom',
 			},
 			removeOnComplete: true,
 			removeOnFail: true,
@@ -60,11 +95,10 @@ export class QueueService {
 			signature,
 		};

-		return this.inboxQueue.add(data, {
+		return this.inboxQueue.add('', data, {
 			attempts: this.config.inboxJobMaxAttempts ?? 8,
-			timeout: 5 * 60 * 1000,	// 5min
 			backoff: {
-				type: 'apBackoff',
+				type: 'custom',
 			},
 			removeOnComplete: true,
 			removeOnFail: true,
@@ -212,7 +246,7 @@ export class QueueService {
 	private generateToDbJobData<T extends 'importFollowingToDb' | 'importBlockingToDb', D extends DbJobData<T>>(name: T, data: D): {
 		name: string,
 		data: D,
-		opts: Bull.JobOptions,
+		opts: Bull.JobsOptions,
 	} {
 		return {
 			name,
@@ -299,10 +333,10 @@ export class QueueService {
 	}

 	@bindThis
-	private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock', data: RelationshipJobData, opts: Bull.JobOptions = {}): {
+	private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock', data: RelationshipJobData, opts: Bull.JobsOptions = {}): {
 		name: string,
 		data: RelationshipJobData,
-		opts: Bull.JobOptions,
+		opts: Bull.JobsOptions,
 	} {
 		return {
 			name,
@@ -351,11 +385,10 @@ export class QueueService {
 			eventId: uuid(),
 		};

-		return this.webhookDeliverQueue.add(data, {
+		return this.webhookDeliverQueue.add(webhook.id, data, {
 			attempts: 4,
-			timeout: 1 * 60 * 1000,	// 1min
 			backoff: {
-				type: 'apBackoff',
+				type: 'custom',
 			},
 			removeOnComplete: true,
 			removeOnFail: true,
@@ -367,11 +400,11 @@ export class QueueService {
 		this.deliverQueue.once('cleaned', (jobs, status) => {
 			//deliverLogger.succ(`Cleaned ${jobs.length} ${status} jobs`);
 		});
-		this.deliverQueue.clean(0, 'delayed');
+		this.deliverQueue.clean(0, Infinity, 'delayed');

 		this.inboxQueue.once('cleaned', (jobs, status) => {
 			//inboxLogger.succ(`Cleaned ${jobs.length} ${status} jobs`);
 		});
-		this.inboxQueue.clean(0, 'delayed');
+		this.inboxQueue.clean(0, Infinity, 'delayed');
 	}
 }
--- a/packages/backend/src/core/activitypub/LdSignatureService.ts
+++ b/packages/backend/src/core/activitypub/LdSignatureService.ts
@@ -94,7 +94,7 @@ class LdSignature {
 	@bindThis
 	private getLoader() {
 		return async (url: string): Promise<any> => {
-			if (!url.match('^https?\:\/\/')) throw `Invalid URL ${url}`;
+			if (!url.match('^https?\:\/\/')) throw new Error(`Invalid URL ${url}`);

 			if (this.preLoad) {
 				if (url in CONTEXTS) {
@@ -126,7 +126,7 @@ class LdSignature {
 			timeout: this.loderTimeout,
 		}, { throwErrorWhenResponseNotOk: false }).then(res => {
 			if (!res.ok) {
-				throw `${res.status} ${res.statusText}`;
+				throw new Error(`${res.status} ${res.statusText}`);
 			} else {
 				return res.json();
 			}
--- a/packages/backend/src/core/activitypub/models/ApNoteService.ts
+++ b/packages/backend/src/core/activitypub/models/ApNoteService.ts
@@ -18,6 +18,7 @@ import { PollService } from '@/core/PollService.js';
 import { StatusError } from '@/misc/status-error.js';
 import { UtilityService } from '@/core/UtilityService.js';
 import { bindThis } from '@/decorators.js';
+import { checkHttps } from '@/misc/check-https.js';
 import { getOneApId, getApId, getOneApHrefNullable, validPost, isEmoji, getApType } from '../type.js';
 // eslint-disable-next-line @typescript-eslint/consistent-type-imports
 import { ApLoggerService } from '../ApLoggerService.js';
@@ -32,7 +33,6 @@ import { ApQuestionService } from './ApQuestionService.js';
 import { ApImageService } from './ApImageService.js';
 import type { Resolver } from '../ApResolverService.js';
 import type { IObject, IPost } from '../type.js';
-import { checkHttps } from '@/misc/check-https.js';

@Injectable()
 export class ApNoteService {
@@ -230,7 +230,7 @@ export class ApNoteService {
 			quote = results.filter((x): x is { status: 'ok', res: Note | null } => x.status === 'ok').map(x => x.res).find(x => x);
 			if (!quote) {
 				if (results.some(x => x.status === 'temperror')) {
-					throw 'quote resolve failed';
+					throw new Error('quote resolve failed');
 				}
 			}
 		}
@@ -311,7 +311,7 @@ export class ApNoteService {
 	
 		// ブロックしてたら中断
 		const meta = await this.metaService.fetch();
-		if (this.utilityService.isBlockedHost(meta.blockedHosts, this.utilityService.extractDbHost(uri))) throw { statusCode: 451 };
+		if (this.utilityService.isBlockedHost(meta.blockedHosts, this.utilityService.extractDbHost(uri))) throw new StatusError('blocked host', 451);
 	
 		const unlock = await this.appLockService.getApLock(uri);