server/controllers/api/jobs.ts

   1 import { Job as BullJob } from 'bullmq'
   2 import express from 'express'
   3 import { HttpStatusCode, Job, JobState, JobType, ResultList, UserRight } from '@shared/models'
   4 import { isArray } from '../../helpers/custom-validators/misc'
   5 import { JobQueue } from '../../lib/job-queue'
   6 import {
   7   asyncMiddleware,
   8   authenticate,
   9   ensureUserHasRight,
  10   jobsSortValidator,
  11   openapiOperationDoc,
  12   paginationValidatorBuilder,
  13   setDefaultPagination,
  14   setDefaultSort
  15 } from '../../middlewares'
  16 import { listJobsValidator } from '../../middlewares/validators/jobs'
  17
  18 const jobsRouter = express.Router()
  19
  20 jobsRouter.post('/pause',
  21   authenticate,
  22   ensureUserHasRight(UserRight.MANAGE_JOBS),
  23   asyncMiddleware(pauseJobQueue)
  24 )
  25
  26 jobsRouter.post('/resume',
  27   authenticate,
  28   ensureUserHasRight(UserRight.MANAGE_JOBS),
  29   resumeJobQueue
  30 )
  31
  32 jobsRouter.get('/:state?',
  33   openapiOperationDoc({ operationId: 'getJobs' }),
  34   authenticate,
  35   ensureUserHasRight(UserRight.MANAGE_JOBS),
  36   paginationValidatorBuilder([ 'jobs' ]),
  37   jobsSortValidator,
  38   setDefaultSort,
  39   setDefaultPagination,
  40   listJobsValidator,
  41   asyncMiddleware(listJobs)
  42 )
  43
  44 // ---------------------------------------------------------------------------
  45
  46 export {
  47   jobsRouter
  48 }
  49
  50 // ---------------------------------------------------------------------------
  51
  52 async function pauseJobQueue (req: express.Request, res: express.Response) {
  53   await JobQueue.Instance.pause()
  54
  55   return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
  56 }
  57
  58 function resumeJobQueue (req: express.Request, res: express.Response) {
  59   JobQueue.Instance.resume()
  60
  61   return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
  62 }
  63
  64 async function listJobs (req: express.Request, res: express.Response) {
  65   const state = req.params.state as JobState
  66   const asc = req.query.sort === 'createdAt'
  67   const jobType = req.query.jobType
  68
  69   const jobs = await JobQueue.Instance.listForApi({
  70     state,
  71     start: req.query.start,
  72     count: req.query.count,
  73     asc,
  74     jobType
  75   })
  76   const total = await JobQueue.Instance.count(state, jobType)
  77
  78   const result: ResultList<Job> = {
  79     total,
  80     data: await Promise.all(jobs.map(j => formatJob(j, state)))
  81   }
  82
  83   return res.json(result)
  84 }
  85
  86 async function formatJob (job: BullJob, state?: JobState): Promise<Job> {
  87   const error = isArray(job.stacktrace) && job.stacktrace.length !== 0
  88     ? job.stacktrace[0]
  89     : null
  90
  91   return {
  92     id: job.id,
  93     state: state || await job.getState(),
  94     type: job.queueName as JobType,
  95     data: job.data,
  96     progress: job.progress as number,
  97     priority: job.opts.priority,
  98     error,
  99     createdAt: new Date(job.timestamp),
 100     finishedOn: new Date(job.finishedOn),
 101     processedOn: new Date(job.processedOn)
 102   }
 103 }