enhance(backend): BullMQの廃止されたRepeatableからJob Schedulersに移行

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