diff --git a/packages/backend/src/GlobalModule.ts b/packages/backend/src/GlobalModule.ts index 7fc8d72d21..d648febb86 100644 --- a/packages/backend/src/GlobalModule.ts +++ b/packages/backend/src/GlobalModule.ts @@ -79,23 +79,11 @@ const $redisForTimelines: Provider = { inject: [DI.config], }; -const $redisForJobQueue: Provider = { - provide: DI.redisForJobQueue, - useFactory: (config: Config) => { - return new Redis.Redis({ - ...config.redisForJobQueue, - maxRetriesPerRequest: null, - keyPrefix: undefined, - }); - }, - inject: [DI.config], -}; - @Global() @Module({ imports: [RepositoryModule], - providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForJobQueue], - exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForJobQueue, RepositoryModule], + providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines], + exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, RepositoryModule], }) export class GlobalModule implements OnApplicationShutdown { constructor( @@ -104,7 +92,6 @@ export class GlobalModule implements OnApplicationShutdown { @Inject(DI.redisForPub) private redisForPub: Redis.Redis, @Inject(DI.redisForSub) private redisForSub: Redis.Redis, @Inject(DI.redisForTimelines) private redisForTimelines: Redis.Redis, - @Inject(DI.redisForJobQueue) private redisForJobQueue: Redis.Redis, ) {} public async dispose(): Promise { @@ -122,7 +109,6 @@ export class GlobalModule implements OnApplicationShutdown { this.redisForPub.disconnect(), this.redisForSub.disconnect(), this.redisForTimelines.disconnect(), - this.redisForJobQueue.disconnect(), ]); } diff --git a/packages/backend/src/core/QueueModule.ts b/packages/backend/src/core/QueueModule.ts index 07a8eb9bfd..4444dc9787 100644 --- a/packages/backend/src/core/QueueModule.ts +++ b/packages/backend/src/core/QueueModule.ts @@ -6,7 +6,6 @@ import { setTimeout } from 'node:timers/promises'; import { Inject, Module, OnApplicationShutdown } from '@nestjs/common'; import * as Bull from 'bullmq'; -import * as Redis from 'ioredis'; import { DI } from '@/di-symbols.js'; import type { Config } from '@/config.js'; import { QUEUE, baseQueueOptions } from '@/queue/const.js'; @@ -24,50 +23,50 @@ export type WebhookDeliverQueue = Bull.Queue; const $system: Provider = { provide: 'queue:system', - useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM)), + inject: [DI.config], }; const $endedPollNotification: Provider = { provide: 'queue:endedPollNotification', - useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.ENDED_POLL_NOTIFICATION, baseQueueOptions(config, QUEUE.ENDED_POLL_NOTIFICATION, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + 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, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + useFactory: (config: Config) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER)), + inject: [DI.config], }; const $inbox: Provider = { provide: 'queue:inbox', - useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + useFactory: (config: Config) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX)), + inject: [DI.config], }; const $db: Provider = { provide: 'queue:db', - useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + useFactory: (config: Config) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB)), + inject: [DI.config], }; const $relationship: Provider = { provide: 'queue:relationship', - useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + useFactory: (config: Config) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP)), + inject: [DI.config], }; const $objectStorage: Provider = { provide: 'queue:objectStorage', - useFactory: (config: Config, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.OBJECT_STORAGE, baseQueueOptions(config, QUEUE.OBJECT_STORAGE, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + 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, redisForJobQueue: Redis.Redis) => new Bull.Queue(QUEUE.WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.WEBHOOK_DELIVER, redisForJobQueue)), - inject: [DI.config, DI.redisForJobQueue], + useFactory: (config: Config) => new Bull.Queue(QUEUE.WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.WEBHOOK_DELIVER)), + inject: [DI.config], }; @Module({ diff --git a/packages/backend/src/daemons/QueueStatsService.ts b/packages/backend/src/daemons/QueueStatsService.ts index 09eba369e0..5edc0f45ab 100644 --- a/packages/backend/src/daemons/QueueStatsService.ts +++ b/packages/backend/src/daemons/QueueStatsService.ts @@ -6,7 +6,6 @@ import { Inject, Injectable } from '@nestjs/common'; import Xev from 'xev'; import * as Bull from 'bullmq'; -import * as Redis from 'ioredis'; import { QueueService } from '@/core/QueueService.js'; import { bindThis } from '@/decorators.js'; import { DI } from '@/di-symbols.js'; @@ -26,9 +25,6 @@ export class QueueStatsService implements OnApplicationShutdown { @Inject(DI.config) private config: Config, - @Inject(DI.redisForJobQueue) - private redisForJobQueue: Redis.Redis, - private queueService: QueueService, ) { } @@ -47,8 +43,8 @@ export class QueueStatsService implements OnApplicationShutdown { let activeDeliverJobs = 0; let activeInboxJobs = 0; - const deliverQueueEvents = new Bull.QueueEvents(QUEUE.DELIVER, baseQueueOptions(this.config, QUEUE.DELIVER, this.redisForJobQueue)); - const inboxQueueEvents = new Bull.QueueEvents(QUEUE.INBOX, baseQueueOptions(this.config, QUEUE.INBOX, this.redisForJobQueue)); + const deliverQueueEvents = new Bull.QueueEvents(QUEUE.DELIVER, baseQueueOptions(this.config, QUEUE.DELIVER)); + const inboxQueueEvents = new Bull.QueueEvents(QUEUE.INBOX, baseQueueOptions(this.config, QUEUE.INBOX)); deliverQueueEvents.on('active', () => { activeDeliverJobs++; diff --git a/packages/backend/src/di-symbols.ts b/packages/backend/src/di-symbols.ts index 410310e159..8403f80659 100644 --- a/packages/backend/src/di-symbols.ts +++ b/packages/backend/src/di-symbols.ts @@ -11,7 +11,6 @@ export const DI = { redisForPub: Symbol('redisForPub'), redisForSub: Symbol('redisForSub'), redisForTimelines: Symbol('redisForTimelines'), - redisForJobQueue: Symbol('redisForJobQueue'), //#region Repositories usersRepository: Symbol('usersRepository'), diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts index 2bec860c00..b5298aa898 100644 --- a/packages/backend/src/queue/QueueProcessorService.ts +++ b/packages/backend/src/queue/QueueProcessorService.ts @@ -5,7 +5,6 @@ import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common'; import * as Bull from 'bullmq'; -import * as Redis from 'ioredis'; import type { Config } from '@/config.js'; import { DI } from '@/di-symbols.js'; import type Logger from '@/logger.js'; @@ -85,9 +84,6 @@ export class QueueProcessorService implements OnApplicationShutdown { @Inject(DI.config) private config: Config, - @Inject(DI.redisForJobQueue) - private redisForJobQueue: Redis.Redis, - private queueLoggerService: QueueLoggerService, private webhookDeliverProcessorService: WebhookDeliverProcessorService, private endedPollNotificationProcessorService: EndedPollNotificationProcessorService, @@ -150,7 +146,7 @@ export class QueueProcessorService implements OnApplicationShutdown { default: throw new Error(`unrecognized job type ${job.name} for system`); } }, { - ...baseQueueOptions(this.config, QUEUE.SYSTEM, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.SYSTEM), autorun: false, }); @@ -189,7 +185,7 @@ export class QueueProcessorService implements OnApplicationShutdown { default: throw new Error(`unrecognized job type ${job.name} for db`); } }, { - ...baseQueueOptions(this.config, QUEUE.DB, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.DB), autorun: false, }); @@ -205,7 +201,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region deliver this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), { - ...baseQueueOptions(this.config, QUEUE.DELIVER, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.DELIVER), autorun: false, concurrency: this.config.deliverJobConcurrency ?? 128, limiter: { @@ -229,7 +225,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region inbox this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), { - ...baseQueueOptions(this.config, QUEUE.INBOX, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.INBOX), autorun: false, concurrency: this.config.inboxJobConcurrency ?? 16, limiter: { @@ -253,7 +249,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region webhook deliver this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => this.webhookDeliverProcessorService.process(job), { - ...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER), autorun: false, concurrency: 64, limiter: { @@ -285,7 +281,7 @@ export class QueueProcessorService implements OnApplicationShutdown { default: throw new Error(`unrecognized job type ${job.name} for relationship`); } }, { - ...baseQueueOptions(this.config, QUEUE.RELATIONSHIP, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.RELATIONSHIP), autorun: false, concurrency: this.config.relashionshipJobConcurrency ?? 16, limiter: { @@ -312,7 +308,7 @@ export class QueueProcessorService implements OnApplicationShutdown { default: throw new Error(`unrecognized job type ${job.name} for objectStorage`); } }, { - ...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE), autorun: false, concurrency: 16, }); @@ -329,7 +325,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region ended poll notification this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => this.endedPollNotificationProcessorService.process(job), { - ...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION, this.redisForJobQueue), + ...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION), autorun: false, }); //#endregion diff --git a/packages/backend/src/queue/const.ts b/packages/backend/src/queue/const.ts index 5b6d59f60c..e6ff7a4b57 100644 --- a/packages/backend/src/queue/const.ts +++ b/packages/backend/src/queue/const.ts @@ -5,7 +5,6 @@ import { Config } from '@/config.js'; import type * as Bull from 'bullmq'; -import type * as Redis from 'ioredis'; export const QUEUE = { DELIVER: 'deliver', @@ -18,9 +17,13 @@ export const QUEUE = { WEBHOOK_DELIVER: 'webhookDeliver', }; -export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE], redisConnection: Redis.Redis): Bull.QueueOptions { +export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE]): Bull.QueueOptions { return { - connection: redisConnection, + connection: { + ...config.redisForJobQueue, + maxRetriesPerRequest: null, + keyPrefix: undefined, + }, prefix: config.redisForJobQueue.prefix ? `${config.redisForJobQueue.prefix}:queue:${queueName}` : `queue:${queueName}`, }; }