From bd911b54b555b11df7e9849cf92d358bccfecf6e Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Mon, 8 Aug 2022 15:48:17 +0200 Subject: Use bullmq job dependency --- .../lib/job-queue/handlers/activitypub-follow.ts | 2 +- .../handlers/activitypub-http-broadcast.ts | 2 +- .../job-queue/handlers/activitypub-http-fetcher.ts | 2 +- .../job-queue/handlers/activitypub-http-unicast.ts | 2 +- .../job-queue/handlers/activitypub-refresher.ts | 2 +- server/lib/job-queue/handlers/actor-keys.ts | 2 +- server/lib/job-queue/handlers/email.ts | 2 +- server/lib/job-queue/handlers/federate-video.ts | 28 ++++++++ .../lib/job-queue/handlers/manage-video-torrent.ts | 2 +- .../job-queue/handlers/move-to-object-storage.ts | 4 +- server/lib/job-queue/handlers/notify.ts | 27 ++++++++ server/lib/job-queue/handlers/video-file-import.ts | 7 +- server/lib/job-queue/handlers/video-import.ts | 15 ++-- server/lib/job-queue/handlers/video-redundancy.ts | 2 +- .../lib/job-queue/handlers/video-studio-edition.ts | 10 ++- server/lib/job-queue/job-queue.ts | 81 ++++++++++++++++++---- 16 files changed, 153 insertions(+), 37 deletions(-) create mode 100644 server/lib/job-queue/handlers/federate-video.ts create mode 100644 server/lib/job-queue/handlers/notify.ts (limited to 'server/lib/job-queue') diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts index 944da5be1..a68c32ba0 100644 --- a/server/lib/job-queue/handlers/activitypub-follow.ts +++ b/server/lib/job-queue/handlers/activitypub-follow.ts @@ -17,7 +17,7 @@ async function processActivityPubFollow (job: Job) { const payload = job.data as ActivitypubFollowPayload const host = payload.host - logger.info('Processing ActivityPub follow in job %d.', job.id) + logger.info('Processing ActivityPub follow in job %s.', job.id) let targetActor: MActorFull if (!host || host === WEBSERVER.HOST) { diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts index 354c608fb..13eff5211 100644 --- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts +++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts @@ -8,7 +8,7 @@ import { doRequest } from '../../../helpers/requests' import { BROADCAST_CONCURRENCY } from '../../../initializers/constants' async function processActivityPubHttpBroadcast (job: Job) { - logger.info('Processing ActivityPub broadcast in job %d.', job.id) + logger.info('Processing ActivityPub broadcast in job %s.', job.id) const payload = job.data as ActivitypubHttpBroadcastPayload diff --git a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts index e0b841887..b6cb3c4a6 100644 --- a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts +++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts @@ -12,7 +12,7 @@ import { addVideoShares } from '../../activitypub/share' import { addVideoComments } from '../../activitypub/video-comments' async function processActivityPubHttpFetcher (job: Job) { - logger.info('Processing ActivityPub fetcher in job %d.', job.id) + logger.info('Processing ActivityPub fetcher in job %s.', job.id) const payload = job.data as ActivitypubHttpFetcherPayload diff --git a/server/lib/job-queue/handlers/activitypub-http-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts index 837a597a5..9e4e84002 100644 --- a/server/lib/job-queue/handlers/activitypub-http-unicast.ts +++ b/server/lib/job-queue/handlers/activitypub-http-unicast.ts @@ -6,7 +6,7 @@ import { doRequest } from '../../../helpers/requests' import { ActorFollowHealthCache } from '../../actor-follow-health-cache' async function processActivityPubHttpUnicast (job: Job) { - logger.info('Processing ActivityPub unicast in job %d.', job.id) + logger.info('Processing ActivityPub unicast in job %s.', job.id) const payload = job.data as ActivitypubHttpUnicastPayload const uri = payload.uri diff --git a/server/lib/job-queue/handlers/activitypub-refresher.ts b/server/lib/job-queue/handlers/activitypub-refresher.ts index 600f858a0..307e771ff 100644 --- a/server/lib/job-queue/handlers/activitypub-refresher.ts +++ b/server/lib/job-queue/handlers/activitypub-refresher.ts @@ -11,7 +11,7 @@ import { refreshActorIfNeeded } from '../../activitypub/actors' async function refreshAPObject (job: Job) { const payload = job.data as RefreshPayload - logger.info('Processing AP refresher in job %d for %s.', job.id, payload.url) + logger.info('Processing AP refresher in job %s for %s.', job.id, payload.url) if (payload.type === 'video') return refreshVideo(payload.url) if (payload.type === 'video-playlist') return refreshVideoPlaylist(payload.url) diff --git a/server/lib/job-queue/handlers/actor-keys.ts b/server/lib/job-queue/handlers/actor-keys.ts index 4a5bad9fb..27a2d431b 100644 --- a/server/lib/job-queue/handlers/actor-keys.ts +++ b/server/lib/job-queue/handlers/actor-keys.ts @@ -6,7 +6,7 @@ import { logger } from '../../../helpers/logger' async function processActorKeys (job: Job) { const payload = job.data as ActorKeysPayload - logger.info('Processing actor keys in job %d.', job.id) + logger.info('Processing actor keys in job %s.', job.id) const actor = await ActorModel.load(payload.actorId) diff --git a/server/lib/job-queue/handlers/email.ts b/server/lib/job-queue/handlers/email.ts index b5b9475b1..567bcc076 100644 --- a/server/lib/job-queue/handlers/email.ts +++ b/server/lib/job-queue/handlers/email.ts @@ -5,7 +5,7 @@ import { Emailer } from '../../emailer' async function processEmail (job: Job) { const payload = job.data as EmailPayload - logger.info('Processing email in job %d.', job.id) + logger.info('Processing email in job %s.', job.id) return Emailer.Instance.sendMail(payload) } diff --git a/server/lib/job-queue/handlers/federate-video.ts b/server/lib/job-queue/handlers/federate-video.ts new file mode 100644 index 000000000..6aac36741 --- /dev/null +++ b/server/lib/job-queue/handlers/federate-video.ts @@ -0,0 +1,28 @@ +import { Job } from 'bullmq' +import { retryTransactionWrapper } from '@server/helpers/database-utils' +import { sequelizeTypescript } from '@server/initializers/database' +import { federateVideoIfNeeded } from '@server/lib/activitypub/videos' +import { VideoModel } from '@server/models/video/video' +import { FederateVideoPayload } from '@shared/models' +import { logger } from '../../../helpers/logger' + +function processFederateVideo (job: Job) { + const payload = job.data as FederateVideoPayload + + logger.info('Processing video federation in job %s.', job.id) + + return retryTransactionWrapper(() => { + return sequelizeTypescript.transaction(async t => { + const video = await VideoModel.loadFull(payload.videoUUID, t) + if (!video) return + + return federateVideoIfNeeded(video, payload.isNewVideo, t) + }) + }) +} + +// --------------------------------------------------------------------------- + +export { + processFederateVideo +} diff --git a/server/lib/job-queue/handlers/manage-video-torrent.ts b/server/lib/job-queue/handlers/manage-video-torrent.ts index 4505ca79e..03aa414c9 100644 --- a/server/lib/job-queue/handlers/manage-video-torrent.ts +++ b/server/lib/job-queue/handlers/manage-video-torrent.ts @@ -8,7 +8,7 @@ import { logger } from '../../../helpers/logger' async function processManageVideoTorrent (job: Job) { const payload = job.data as ManageVideoTorrentPayload - logger.info('Processing torrent in job %d.', job.id) + logger.info('Processing torrent in job %s.', job.id) if (payload.action === 'create') return doCreateAction(payload) if (payload.action === 'update-metadata') return doUpdateMetadataAction(payload) diff --git a/server/lib/job-queue/handlers/move-to-object-storage.ts b/server/lib/job-queue/handlers/move-to-object-storage.ts index d608fd865..25bdebeea 100644 --- a/server/lib/job-queue/handlers/move-to-object-storage.ts +++ b/server/lib/job-queue/handlers/move-to-object-storage.ts @@ -17,7 +17,7 @@ const lTagsBase = loggerTagsFactory('move-object-storage') export async function processMoveToObjectStorage (job: Job) { const payload = job.data as MoveObjectStoragePayload - logger.info('Moving video %s in job %d.', payload.videoUUID, job.id) + logger.info('Moving video %s in job %s.', payload.videoUUID, job.id) const video = await VideoModel.loadWithFiles(payload.videoUUID) // No video, maybe deleted? @@ -43,7 +43,7 @@ export async function processMoveToObjectStorage (job: Job) { const pendingMove = await VideoJobInfoModel.decrease(video.uuid, 'pendingMove') if (pendingMove === 0) { - logger.info('Running cleanup after moving files to object storage (video %s in job %d)', video.uuid, job.id, lTags) + logger.info('Running cleanup after moving files to object storage (video %s in job %s)', video.uuid, job.id, lTags) await doAfterLastJob({ video, previousVideoState: payload.previousVideoState, isNewVideo: payload.isNewVideo }) } diff --git a/server/lib/job-queue/handlers/notify.ts b/server/lib/job-queue/handlers/notify.ts new file mode 100644 index 000000000..83605396c --- /dev/null +++ b/server/lib/job-queue/handlers/notify.ts @@ -0,0 +1,27 @@ +import { Job } from 'bullmq' +import { Notifier } from '@server/lib/notifier' +import { VideoModel } from '@server/models/video/video' +import { NotifyPayload } from '@shared/models' +import { logger } from '../../../helpers/logger' + +async function processNotify (job: Job) { + const payload = job.data as NotifyPayload + logger.info('Processing %s notification in job %s.', payload.action, job.id) + + if (payload.action === 'new-video') return doNotifyNewVideo(payload) +} + +// --------------------------------------------------------------------------- + +export { + processNotify +} + +// --------------------------------------------------------------------------- + +async function doNotifyNewVideo (payload: NotifyPayload & { action: 'new-video' }) { + const refreshedVideo = await VideoModel.loadFull(payload.videoUUID) + if (!refreshedVideo) return + + Notifier.Instance.notifyOnNewVideoIfNeeded(refreshedVideo) +} diff --git a/server/lib/job-queue/handlers/video-file-import.ts b/server/lib/job-queue/handlers/video-file-import.ts index 40c44cf52..d950f6407 100644 --- a/server/lib/job-queue/handlers/video-file-import.ts +++ b/server/lib/job-queue/handlers/video-file-import.ts @@ -4,7 +4,7 @@ import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent' import { CONFIG } from '@server/initializers/config' import { federateVideoIfNeeded } from '@server/lib/activitypub/videos' import { generateWebTorrentVideoFilename } from '@server/lib/paths' -import { addMoveToObjectStorageJob } from '@server/lib/video' +import { buildMoveToObjectStorageJob } from '@server/lib/video' import { VideoPathManager } from '@server/lib/video-path-manager' import { VideoModel } from '@server/models/video/video' import { VideoFileModel } from '@server/models/video/video-file' @@ -13,10 +13,11 @@ import { getLowercaseExtension } from '@shared/core-utils' import { VideoFileImportPayload, VideoStorage } from '@shared/models' import { getVideoStreamFPS, getVideoStreamDimensionsInfo } from '../../../helpers/ffmpeg' import { logger } from '../../../helpers/logger' +import { JobQueue } from '../job-queue' async function processVideoFileImport (job: Job) { const payload = job.data as VideoFileImportPayload - logger.info('Processing video file import in job %d.', job.id) + logger.info('Processing video file import in job %s.', job.id) const video = await VideoModel.loadFull(payload.videoUUID) // No video, maybe deleted? @@ -28,7 +29,7 @@ async function processVideoFileImport (job: Job) { await updateVideoFile(video, payload.filePath) if (CONFIG.OBJECT_STORAGE.ENABLED) { - await addMoveToObjectStorageJob({ video, previousVideoState: video.state }) + await JobQueue.Instance.createJob(await buildMoveToObjectStorageJob({ video, previousVideoState: video.state })) } else { await federateVideoIfNeeded(video, false) } diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts index e5cd35865..f4629159c 100644 --- a/server/lib/job-queue/handlers/video-import.ts +++ b/server/lib/job-queue/handlers/video-import.ts @@ -8,7 +8,7 @@ import { generateWebTorrentVideoFilename } from '@server/lib/paths' import { Hooks } from '@server/lib/plugins/hooks' import { ServerConfigManager } from '@server/lib/server-config-manager' import { isAbleToUploadVideo } from '@server/lib/user' -import { addMoveToObjectStorageJob, addOptimizeOrMergeAudioJob } from '@server/lib/video' +import { buildOptimizeOrMergeAudioJob, buildMoveToObjectStorageJob } from '@server/lib/video' import { VideoPathManager } from '@server/lib/video-path-manager' import { buildNextVideoState } from '@server/lib/video-state' import { ThumbnailModel } from '@server/models/video/thumbnail' @@ -39,6 +39,7 @@ import { MThumbnail } from '../../../types/models/video/thumbnail' import { federateVideoIfNeeded } from '../../activitypub/videos' import { Notifier } from '../../notifier' import { generateVideoMiniature } from '../../thumbnail' +import { JobQueue } from '../job-queue' async function processVideoImport (job: Job) { const payload = job.data as VideoImportPayload @@ -65,7 +66,7 @@ export { // --------------------------------------------------------------------------- async function processTorrentImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportTorrentPayload) { - logger.info('Processing torrent video import in job %d.', job.id) + logger.info('Processing torrent video import in job %s.', job.id) const options = { type: payload.type, videoImportId: payload.videoImportId } @@ -77,7 +78,7 @@ async function processTorrentImport (job: Job, videoImport: MVideoImportDefault, } async function processYoutubeDLImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportYoutubeDLPayload) { - logger.info('Processing youtubeDL video import in job %d.', job.id) + logger.info('Processing youtubeDL video import in job %s.', job.id) const options = { type: payload.type, videoImportId: videoImport.id } @@ -259,12 +260,16 @@ async function processFile (downloader: () => Promise, videoImport: MVid } if (video.state === VideoState.TO_MOVE_TO_EXTERNAL_STORAGE) { - return addMoveToObjectStorageJob({ video: videoImportUpdated.Video, previousVideoState: VideoState.TO_IMPORT }) + await JobQueue.Instance.createJob( + await buildMoveToObjectStorageJob({ video: videoImportUpdated.Video, previousVideoState: VideoState.TO_IMPORT }) + ) } // Create transcoding jobs? if (video.state === VideoState.TO_TRANSCODE) { - await addOptimizeOrMergeAudioJob({ video: videoImportUpdated.Video, videoFile, user: videoImport.User }) + await JobQueue.Instance.createJob( + await buildOptimizeOrMergeAudioJob({ video: videoImportUpdated.Video, videoFile, user: videoImport.User }) + ) } } catch (err) { diff --git a/server/lib/job-queue/handlers/video-redundancy.ts b/server/lib/job-queue/handlers/video-redundancy.ts index 75ab2cd02..bac99fdb7 100644 --- a/server/lib/job-queue/handlers/video-redundancy.ts +++ b/server/lib/job-queue/handlers/video-redundancy.ts @@ -5,7 +5,7 @@ import { logger } from '../../../helpers/logger' async function processVideoRedundancy (job: Job) { const payload = job.data as VideoRedundancyPayload - logger.info('Processing video redundancy in job %d.', job.id) + logger.info('Processing video redundancy in job %s.', job.id) return VideosRedundancyScheduler.Instance.createManualRedundancy(payload.videoId) } diff --git a/server/lib/job-queue/handlers/video-studio-edition.ts b/server/lib/job-queue/handlers/video-studio-edition.ts index 078243538..23f9a34cc 100644 --- a/server/lib/job-queue/handlers/video-studio-edition.ts +++ b/server/lib/job-queue/handlers/video-studio-edition.ts @@ -8,7 +8,7 @@ import { federateVideoIfNeeded } from '@server/lib/activitypub/videos' import { generateWebTorrentVideoFilename } from '@server/lib/paths' import { VideoTranscodingProfilesManager } from '@server/lib/transcoding/default-transcoding-profiles' import { isAbleToUploadVideo } from '@server/lib/user' -import { addOptimizeOrMergeAudioJob } from '@server/lib/video' +import { buildOptimizeOrMergeAudioJob } from '@server/lib/video' import { removeHLSPlaylist, removeWebTorrentFile } from '@server/lib/video-file' import { VideoPathManager } from '@server/lib/video-path-manager' import { approximateIntroOutroAdditionalSize } from '@server/lib/video-studio' @@ -36,6 +36,7 @@ import { VideoStudioTaskWatermarkPayload } from '@shared/models' import { logger, loggerTagsFactory } from '../../../helpers/logger' +import { JobQueue } from '../job-queue' const lTagsBase = loggerTagsFactory('video-edition') @@ -43,7 +44,7 @@ async function processVideoStudioEdition (job: Job) { const payload = job.data as VideoStudioEditionPayload const lTags = lTagsBase(payload.videoUUID) - logger.info('Process video studio edition of %s in job %d.', payload.videoUUID, job.id, lTags) + logger.info('Process video studio edition of %s in job %s.', payload.videoUUID, job.id, lTags) const video = await VideoModel.loadFull(payload.videoUUID) @@ -100,7 +101,10 @@ async function processVideoStudioEdition (job: Job) { await federateVideoIfNeeded(video, false, undefined) const user = await UserModel.loadByVideoId(video.id) - await addOptimizeOrMergeAudioJob({ video, videoFile: newFile, user, isNewVideo: false }) + + await JobQueue.Instance.createJob( + await buildOptimizeOrMergeAudioJob({ video, videoFile: newFile, user, isNewVideo: false }) + ) } // --------------------------------------------------------------------------- diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index 0cf5d53ce..50d732beb 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -1,4 +1,6 @@ import { + FlowJob, + FlowProducer, Job, JobsOptions, Queue, @@ -13,7 +15,7 @@ import { import { jobStates } from '@server/helpers/custom-validators/jobs' import { CONFIG } from '@server/initializers/config' import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy' -import { timeoutPromise } from '@shared/core-utils' +import { pick, timeoutPromise } from '@shared/core-utils' import { ActivitypubFollowPayload, ActivitypubHttpBroadcastPayload, @@ -22,10 +24,12 @@ import { ActorKeysPayload, DeleteResumableUploadMetaFilePayload, EmailPayload, + FederateVideoPayload, JobState, JobType, ManageVideoTorrentPayload, MoveObjectStoragePayload, + NotifyPayload, RefreshPayload, VideoFileImportPayload, VideoImportPayload, @@ -45,8 +49,10 @@ import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unica import { refreshAPObject } from './handlers/activitypub-refresher' import { processActorKeys } from './handlers/actor-keys' import { processEmail } from './handlers/email' +import { processFederateVideo } from './handlers/federate-video' import { processManageVideoTorrent } from './handlers/manage-video-torrent' import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage' +import { processNotify } from './handlers/notify' import { processVideoFileImport } from './handlers/video-file-import' import { processVideoImport } from './handlers/video-import' import { processVideoLiveEnding } from './handlers/video-live-ending' @@ -54,7 +60,7 @@ import { processVideoStudioEdition } from './handlers/video-studio-edition' import { processVideoTranscoding } from './handlers/video-transcoding' import { processVideosViewsStats } from './handlers/video-views-stats' -type CreateJobArgument = +export type CreateJobArgument = { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | { type: 'activitypub-http-broadcast-parallel', payload: ActivitypubHttpBroadcastPayload } | { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | @@ -73,7 +79,9 @@ type CreateJobArgument = { type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } | { type: 'video-studio-edition', payload: VideoStudioEditionPayload } | { type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } | - { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } + { type: 'notify', payload: NotifyPayload } | + { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } | + { type: 'federate-video', payload: FederateVideoPayload } export type CreateJobOptions = { delay?: number @@ -98,7 +106,9 @@ const handlers: { [id in JobType]: (job: Job) => Promise } = { 'video-redundancy': processVideoRedundancy, 'move-to-object-storage': processMoveToObjectStorage, 'manage-video-torrent': processManageVideoTorrent, - 'video-studio-edition': processVideoStudioEdition + 'notify': processNotify, + 'video-studio-edition': processVideoStudioEdition, + 'federate-video': processFederateVideo } const errorHandlers: { [id in JobType]?: (job: Job, err: any) => Promise } = { @@ -123,7 +133,9 @@ const jobTypes: JobType[] = [ 'video-live-ending', 'move-to-object-storage', 'manage-video-torrent', - 'video-studio-edition' + 'video-studio-edition', + 'notify', + 'federate-video' ] const silentFailure = new Set([ 'activitypub-http-unicast' ]) @@ -137,6 +149,8 @@ class JobQueue { private queueSchedulers: { [id in JobType]?: QueueScheduler } = {} private queueEvents: { [id in JobType]?: QueueEvents } = {} + private flowProducer: FlowProducer + private initialized = false private jobRedisPrefix: string @@ -157,6 +171,11 @@ class JobQueue { this.buildQueueEvent(handlerName, produceOnly) } + this.flowProducer = new FlowProducer({ + connection: this.getRedisConnection(), + prefix: this.jobRedisPrefix + }) + this.addRepeatableJobs() } @@ -243,6 +262,8 @@ class JobQueue { } } + // --------------------------------------------------------------------------- + async terminate () { const promises = Object.keys(this.workers) .map(handlerName => { @@ -278,28 +299,56 @@ class JobQueue { } } - createJob (obj: CreateJobArgument, options: CreateJobOptions = {}): void { - this.createJobWithPromise(obj, options) - .catch(err => logger.error('Cannot create job.', { err, obj })) + // --------------------------------------------------------------------------- + + createJobAsync (options: CreateJobArgument & CreateJobOptions): void { + this.createJob(options) + .catch(err => logger.error('Cannot create job.', { err, options })) } - async createJobWithPromise (obj: CreateJobArgument, options: CreateJobOptions = {}) { - const queue: Queue = this.queues[obj.type] + async createJob (options: CreateJobArgument & CreateJobOptions) { + const queue: Queue = this.queues[options.type] if (queue === undefined) { - logger.error('Unknown queue %s: cannot create job.', obj.type) + logger.error('Unknown queue %s: cannot create job.', options.type) return } - const jobArgs: JobsOptions = { + const jobOptions = this.buildJobOptions(options.type as JobType, pick(options, [ 'priority', 'delay' ])) + + return queue.add('job', options.payload, jobOptions) + } + + async createSequentialJobFlow (...jobs: ((CreateJobArgument & CreateJobOptions) | undefined)[]) { + let lastJob: FlowJob + + for (const job of jobs) { + if (!job) continue + + lastJob = { + name: 'job', + data: job.payload, + queueName: job.type, + opts: this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay' ])), + children: lastJob + ? [ lastJob ] + : [] + } + } + + return this.flowProducer.add(lastJob) + } + + private buildJobOptions (type: JobType, options: CreateJobOptions = {}): JobsOptions { + return { backoff: { delay: 60 * 1000, type: 'exponential' }, - attempts: JOB_ATTEMPTS[obj.type], + attempts: JOB_ATTEMPTS[type], priority: options.priority, delay: options.delay } - - return queue.add('job', obj.payload, jobArgs) } + // --------------------------------------------------------------------------- + async listForApi (options: { state?: JobState start: number @@ -367,6 +416,8 @@ class JobQueue { return Promise.all(promises) } + // --------------------------------------------------------------------------- + async removeOldJobs () { for (const key of Object.keys(this.queues)) { const queue: Queue = this.queues[key] -- cgit v1.2.3