From 35888eb8f49332d4f4e4cc38a4bd8a9b3575345a Mon Sep 17 00:00:00 2001
From: syuilo <4439005+syuilo@users.noreply.github.com>
Date: Thu, 31 Jul 2025 18:16:21 +0900
Subject: [PATCH] =?UTF-8?q?enhance(backend):=20BullMQ=E3=81=AE=E5=BB=83?=
 =?UTF-8?q?=E6=AD=A2=E3=81=95=E3=82=8C=E3=81=9FRepeatable=E3=81=8B?=
 =?UTF-8?q?=E3=82=89Job=20Schedulers=E3=81=AB=E7=A7=BB=E8=A1=8C?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 packages/backend/src/core/QueueService.ts | 92 ++++++++++++++---------
 1 file changed, 58 insertions(+), 34 deletions(-)

diff --git a/packages/backend/src/core/QueueService.ts b/packages/backend/src/core/QueueService.ts
index 04bbc7e38a..fce8cc2067 100644
--- a/packages/backend/src/core/QueueService.ts
+++ b/packages/backend/src/core/QueueService.ts
@@ -17,6 +17,7 @@ import { bindThis } from '@/decorators.js';
 import type { Antenna } from '@/server/api/endpoints/i/import-antennas.js';
 import { ApRequestCreator } from '@/core/activitypub/ApRequestService.js';
 import { type SystemWebhookPayload } from '@/core/SystemWebhookService.js';
+import type { Packed } from '@/misc/json-schema.js';
 import { type UserWebhookPayload } from './UserWebhookService.js';
 import type {
 	DbJobData,
@@ -39,7 +40,6 @@ import type {
 } from './QueueModule.js';
 import type httpSignature from '@peertube/http-signature';
 import type * as Bull from 'bullmq';
-import type { Packed } from '@/misc/json-schema.js';
 
 export const QUEUE_TYPES = [
 	'system',
@@ -69,61 +69,85 @@ export class QueueService {
 		@Inject('queue:userWebhookDeliver') public userWebhookDeliverQueue: UserWebhookDeliverQueue,
 		@Inject('queue:systemWebhookDeliver') public systemWebhookDeliverQueue: SystemWebhookDeliverQueue,
 	) {
-		this.systemQueue.add('tickCharts', {
+		this.systemQueue.upsertJobScheduler('tickCharts', {
+			pattern: '55 * * * *',
 		}, {
-			repeat: { pattern: '55 * * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'tickCharts',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('resyncCharts', {
+		this.systemQueue.upsertJobScheduler('resyncCharts', {
+			pattern: '0 0 * * *',
 		}, {
-			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'resyncCharts',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('cleanCharts', {
+		this.systemQueue.upsertJobScheduler('cleanCharts', {
+			pattern: '0 0 * * *',
 		}, {
-			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'cleanCharts',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('aggregateRetention', {
+		this.systemQueue.upsertJobScheduler('aggregateRetention', {
+			pattern: '0 0 * * *',
 		}, {
-			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'aggregateRetention',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('clean', {
+		this.systemQueue.upsertJobScheduler('clean', {
+			pattern: '0 0 * * *',
 		}, {
-			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'clean',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('checkExpiredMutings', {
+		this.systemQueue.upsertJobScheduler('checkExpiredMutings', {
+			pattern: '*/5 * * * *',
 		}, {
-			repeat: { pattern: '*/5 * * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'checkExpiredMutings',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('bakeBufferedReactions', {
+		this.systemQueue.upsertJobScheduler('bakeBufferedReactions', {
+			pattern: '0 0 * * *',
 		}, {
-			repeat: { pattern: '0 0 * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			name: 'bakeBufferedReactions',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 
-		this.systemQueue.add('checkModeratorsActivity', {
-		}, {
+		this.systemQueue.upsertJobScheduler('checkModeratorsActivity', {
 			// 毎時30分に起動
-			repeat: { pattern: '30 * * * *' },
-			removeOnComplete: 10,
-			removeOnFail: 30,
+			pattern: '30 * * * *',
+		}, {
+			name: 'checkModeratorsActivity',
+			opts: {
+				removeOnComplete: 10,
+				removeOnFail: 30,
+			},
 		});
 	}