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