1 files changed, 427 insertions, 71 deletions

diff --git a/packages/backend/src/core/QueueService.ts b/packages/backend/src/core/QueueService.ts
index 039c47724b..fb0fa8f28d 100644
--- a/packages/backend/src/core/QueueService.ts
+++ b/packages/backend/src/core/QueueService.ts
@@ -5,6 +5,8 @@
 
 import { randomUUID } from 'node:crypto';
 import { Inject, Injectable } from '@nestjs/common';
+import { MetricsTime, type JobType } from 'bullmq';
+import { parse as parseRedisInfo } from 'redis-info';
 import type { IActivity } from '@/core/activitypub/type.js';
 import type { MiDriveFile } from '@/models/DriveFile.js';
 import type { MiWebhook, WebhookEventTypes } from '@/models/Webhook.js';
@@ -40,6 +42,18 @@ import type httpSignature from '@peertube/http-signature';
 import type * as Bull from 'bullmq';
 import { MiNote } from '@/models/Note.js';
 
+export const QUEUE_TYPES = [
+	'system',
+	'endedPollNotification',
+	'deliver',
+	'inbox',
+	'db',
+	'relationship',
+	'objectStorage',
+	'userWebhookDeliver',
+	'systemWebhookDeliver',
+] as const;
+
 @Injectable()
 export class QueueService {
 	constructor(
@@ -60,50 +74,58 @@ export class QueueService {
 		this.systemQueue.add('tickCharts', {
 		}, {
 			repeat: { pattern: '55 * * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('resyncCharts', {
 		}, {
 			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('cleanCharts', {
 		}, {
 			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('aggregateRetention', {
 		}, {
 			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('clean', {
 		}, {
 			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('checkExpiredMutings', {
 		}, {
 			repeat: { pattern: '*/5 * * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('bakeBufferedReactions', {
 		}, {
 			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 
 		this.systemQueue.add('checkModeratorsActivity', {
 		}, {
 			// 毎時30分に起動
 			repeat: { pattern: '30 * * * *' },
-			removeOnComplete: true,
+			removeOnComplete: 10,
+			removeOnFail: 30,
 		});
 	}
 
@@ -125,13 +147,21 @@ export class QueueService {
 			isSharedInbox,
 		};
 
-		return this.deliverQueue.add(to, data, {
+		const label = to.replace('https://', '').replace('/inbox', '');
+
+		return this.deliverQueue.add(label, data, {
 			attempts: this.config.deliverJobMaxAttempts ?? 12,
 			backoff: {
 				type: 'custom',
 			},
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -153,12 +183,18 @@ export class QueueService {
 			backoff: {
 				type: 'custom',
 			},
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		};
 
 		await this.deliverQueue.addBulk(Array.from(inboxes.entries(), d => ({
-			name: d[0],
+			name: d[0].replace('https://', '').replace('/inbox', ''),
 			data: {
 				user,
 				content: contentBody,
@@ -179,13 +215,21 @@ export class QueueService {
 			signature,
 		};
 
-		return this.inboxQueue.add('', data, {
+		const label = (activity.id ?? '').replace('https://', '').replace('/activity', '');
+
+		return this.inboxQueue.add(label, data, {
 			attempts: this.config.inboxJobMaxAttempts ?? 8,
 			backoff: {
 				type: 'custom',
 			},
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -194,8 +238,14 @@ export class QueueService {
 		return this.dbQueue.add('deleteDriveFiles', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -204,8 +254,14 @@ export class QueueService {
 		return this.dbQueue.add('exportCustomEmojis', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -224,8 +280,14 @@ export class QueueService {
 		return this.dbQueue.add('exportNotes', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -234,8 +296,14 @@ export class QueueService {
 		return this.dbQueue.add('exportClips', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -244,8 +312,14 @@ export class QueueService {
 		return this.dbQueue.add('exportFavorites', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -256,8 +330,14 @@ export class QueueService {
 			excludeMuting,
 			excludeInactive,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -266,8 +346,14 @@ export class QueueService {
 		return this.dbQueue.add('exportMuting', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -276,8 +362,14 @@ export class QueueService {
 		return this.dbQueue.add('exportBlocking', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -286,8 +378,14 @@ export class QueueService {
 		return this.dbQueue.add('exportUserLists', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -296,8 +394,14 @@ export class QueueService {
 		return this.dbQueue.add('exportAntennas', {
 			user: { id: user.id },
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -308,8 +412,14 @@ export class QueueService {
 			fileId: fileId,
 			withReplies,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -373,8 +483,14 @@ export class QueueService {
 			user: { id: user.id },
 			fileId: fileId,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -384,8 +500,14 @@ export class QueueService {
 			user: { id: user.id },
 			fileId: fileId,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -405,8 +527,14 @@ export class QueueService {
 			name,
 			data,
 			opts: {
-				removeOnComplete: true,
-				removeOnFail: true,
+				removeOnComplete: {
+					age: 3600 * 24 * 7, // keep up to 7 days
+					count: 30,
+				},
+				removeOnFail: {
+					age: 3600 * 24 * 7, // keep up to 7 days
+					count: 100,
+				},
 			},
 		};
 	}
@@ -417,8 +545,14 @@ export class QueueService {
 			user: { id: user.id },
 			fileId: fileId,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -428,8 +562,14 @@ export class QueueService {
 			user: { id: user.id },
 			fileId: fileId,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -439,8 +579,14 @@ export class QueueService {
 			user: { id: user.id },
 			antenna,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -450,8 +596,14 @@ export class QueueService {
 			user: { id: user.id },
 			soft: opts.soft,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -501,8 +653,14 @@ export class QueueService {
 				withReplies: data.withReplies,
 			},
 			opts: {
-				removeOnComplete: true,
-				removeOnFail: true,
+				removeOnComplete: {
+					age: 3600 * 24 * 7, // keep up to 7 days
+					count: 30,
+				},
+				removeOnFail: {
+					age: 3600 * 24 * 7, // keep up to 7 days
+					count: 100,
+				},
 				...opts,
 			},
 		};
@@ -513,16 +671,28 @@ export class QueueService {
 		return this.objectStorageQueue.add('deleteFile', {
 			key: key,
 		}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
 	@bindThis
 	public createCleanRemoteFilesJob() {
 		return this.objectStorageQueue.add('cleanRemoteFiles', {}, {
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -553,8 +723,14 @@ export class QueueService {
 			backoff: {
 				type: 'custom',
 			},
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
@@ -584,21 +760,201 @@ export class QueueService {
 			backoff: {
 				type: 'custom',
 			},
-			removeOnComplete: true,
-			removeOnFail: true,
+			removeOnComplete: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 30,
+			},
+			removeOnFail: {
+				age: 3600 * 24 * 7, // keep up to 7 days
+				count: 100,
+			},
 		});
 	}
 
 	@bindThis
-	public destroy() {
-		this.deliverQueue.once('cleaned', (jobs, status) => {
-			//deliverLogger.succ(`Cleaned ${jobs.length} ${status} jobs`);
-		});
-		this.deliverQueue.clean(0, 0, 'delayed');
+	private getQueue(type: typeof QUEUE_TYPES[number]): Bull.Queue {
+		switch (type) {
+			case 'system': return this.systemQueue;
+			case 'endedPollNotification': return this.endedPollNotificationQueue;
+			case 'deliver': return this.deliverQueue;
+			case 'inbox': return this.inboxQueue;
+			case 'db': return this.dbQueue;
+			case 'relationship': return this.relationshipQueue;
+			case 'objectStorage': return this.objectStorageQueue;
+			case 'userWebhookDeliver': return this.userWebhookDeliverQueue;
+			case 'systemWebhookDeliver': return this.systemWebhookDeliverQueue;
+			default: throw new Error(`Unrecognized queue type: ${type}`);
+		}
+	}
+
+	@bindThis
+	public async queueClear(queueType: typeof QUEUE_TYPES[number], state: '*' | 'completed' | 'wait' | 'active' | 'paused' | 'prioritized' | 'delayed' | 'failed') {
+		const queue = this.getQueue(queueType);
+
+		if (state === '*') {
+			await Promise.all([
+				queue.clean(0, 0, 'completed'),
+				queue.clean(0, 0, 'wait'),
+				queue.clean(0, 0, 'active'),
+				queue.clean(0, 0, 'paused'),
+				queue.clean(0, 0, 'prioritized'),
+				queue.clean(0, 0, 'delayed'),
+				queue.clean(0, 0, 'failed'),
+			]);
+		} else {
+			await queue.clean(0, 0, state);
+		}
+	}
 
-		this.inboxQueue.once('cleaned', (jobs, status) => {
-			//inboxLogger.succ(`Cleaned ${jobs.length} ${status} jobs`);
+	@bindThis
+	public async queuePromoteJobs(queueType: typeof QUEUE_TYPES[number]) {
+		const queue = this.getQueue(queueType);
+		await queue.promoteJobs();
+	}
+
+	@bindThis
+	public async queueRetryJob(queueType: typeof QUEUE_TYPES[number], jobId: string) {
+		const queue = this.getQueue(queueType);
+		const job: Bull.Job | null = await queue.getJob(jobId);
+		if (job) {
+			if (job.finishedOn != null) {
+				await job.retry();
+			} else {
+				await job.promote();
+			}
+		}
+	}
+
+	@bindThis
+	public async queueRemoveJob(queueType: typeof QUEUE_TYPES[number], jobId: string) {
+		const queue = this.getQueue(queueType);
+		const job: Bull.Job | null = await queue.getJob(jobId);
+		if (job) {
+			await job.remove();
+		}
+	}
+
+	@bindThis
+	private packJobData(job: Bull.Job) {
+		// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+		const stacktrace = job.stacktrace ? job.stacktrace.filter(Boolean) : [];
+		stacktrace.reverse();
+
+		return {
+			id: job.id,
+			name: job.name,
+			data: job.data,
+			opts: job.opts,
+			timestamp: job.timestamp,
+			processedOn: job.processedOn,
+			processedBy: job.processedBy,
+			finishedOn: job.finishedOn,
+			progress: job.progress,
+			attempts: job.attemptsMade,
+			delay: job.delay,
+			failedReason: job.failedReason,
+			stacktrace: stacktrace,
+			returnValue: job.returnvalue,
+			isFailed: !!job.failedReason || (Array.isArray(stacktrace) && stacktrace.length > 0),
+		};
+	}
+
+	@bindThis
+	public async queueGetJob(queueType: typeof QUEUE_TYPES[number], jobId: string) {
+		const queue = this.getQueue(queueType);
+		const job: Bull.Job | null = await queue.getJob(jobId);
+		if (job) {
+			return this.packJobData(job);
+		} else {
+			throw new Error(`Job not found: ${jobId}`);
+		}
+	}
+
+	@bindThis
+	public async queueGetJobs(queueType: typeof QUEUE_TYPES[number], jobTypes: JobType[], search?: string) {
+		const RETURN_LIMIT = 100;
+		const queue = this.getQueue(queueType);
+		let jobs: Bull.Job[];
+
+		if (search) {
+			jobs = await queue.getJobs(jobTypes, 0, 1000);
+
+			jobs = jobs.filter(job => {
+				const jobString = JSON.stringify(job).toLowerCase();
+				return search.toLowerCase().split(' ').every(term => {
+					return jobString.includes(term);
+				});
+			});
+
+			jobs = jobs.slice(0, RETURN_LIMIT);
+		} else {
+			jobs = await queue.getJobs(jobTypes, 0, RETURN_LIMIT);
+		}
+
+		return jobs.map(job => this.packJobData(job));
+	}
+
+	@bindThis
+	public async queueGetQueues() {
+		const fetchings = QUEUE_TYPES.map(async type => {
+			const queue = this.getQueue(type);
+
+			const counts = await queue.getJobCounts();
+			const isPaused = await queue.isPaused();
+			const metrics_completed = await queue.getMetrics('completed', 0, MetricsTime.ONE_WEEK);
+			const metrics_failed = await queue.getMetrics('failed', 0, MetricsTime.ONE_WEEK);
+
+			return {
+				name: type,
+				counts: counts,
+				isPaused,
+				metrics: {
+					completed: metrics_completed,
+					failed: metrics_failed,
+				},
+			};
 		});
-		this.inboxQueue.clean(0, 0, 'delayed');
+
+		return await Promise.all(fetchings);
+	}
+
+	@bindThis
+	public async queueGetQueue(queueType: typeof QUEUE_TYPES[number]) {
+		const queue = this.getQueue(queueType);
+		const counts = await queue.getJobCounts();
+		const isPaused = await queue.isPaused();
+		const metrics_completed = await queue.getMetrics('completed', 0, MetricsTime.ONE_WEEK);
+		const metrics_failed = await queue.getMetrics('failed', 0, MetricsTime.ONE_WEEK);
+		const db = parseRedisInfo(await (await queue.client).info());
+
+		return {
+			name: queueType,
+			qualifiedName: queue.qualifiedName,
+			counts: counts,
+			isPaused,
+			metrics: {
+				completed: metrics_completed,
+				failed: metrics_failed,
+			},
+			db: {
+				version: db.redis_version,
+				mode: db.redis_mode,
+				runId: db.run_id,
+				processId: db.process_id,
+				port: parseInt(db.tcp_port),
+				os: db.os,
+				uptime: parseInt(db.uptime_in_seconds),
+				memory: {
+					total: parseInt(db.total_system_memory) || parseInt(db.maxmemory),
+					used: parseInt(db.used_memory),
+					fragmentationRatio: parseInt(db.mem_fragmentation_ratio),
+					peak: parseInt(db.used_memory_peak),
+				},
+				clients: {
+					connected: parseInt(db.connected_clients),
+					blocked: parseInt(db.blocked_clients),
+				},
+			},
+		};
 	}
 }