Add ability to filter per job type

8 files changed, 113 insertions, 26 deletions

diff --git a/server/controllers/api/jobs.ts b/server/controllers/api/jobs.ts
index 1fa662349..05320311e 100644
--- a/server/controllers/api/jobs.ts
+++ b/server/controllers/api/jobs.ts
@@ -24,7 +24,7 @@ jobsRouter.get('/:state',
   jobsSortValidator,
   setDefaultSort,
   setDefaultPagination,
-  asyncMiddleware(listJobsValidator),
+  listJobsValidator,
   asyncMiddleware(listJobs)
 )
 
@@ -39,8 +39,15 @@ export {
 async function listJobs (req: express.Request, res: express.Response) {
   const state = req.params.state as JobState
   const asc = req.query.sort === 'createdAt'
+  const jobType = req.query.jobType
 
-  const jobs = await JobQueue.Instance.listForApi(state, req.query.start, req.query.count, asc)
+  const jobs = await JobQueue.Instance.listForApi({
+    state,
+    start: req.query.start,
+    count: req.query.count,
+    asc,
+    jobType
+  })
   const total = await JobQueue.Instance.count(state)
 
   const result: ResultList<any> = {
diff --git a/server/helpers/custom-validators/jobs.ts b/server/helpers/custom-validators/jobs.ts
index 1cc6e6912..dd33e85a3 100644
--- a/server/helpers/custom-validators/jobs.ts
+++ b/server/helpers/custom-validators/jobs.ts
@@ -1,14 +1,20 @@
 import { JobState } from '../../../shared/models'
 import { exists } from './misc'
+import { jobTypes } from '@server/lib/job-queue/job-queue'
 
 const jobStates: JobState[] = [ 'active', 'completed', 'failed', 'waiting', 'delayed' ]
 
 function isValidJobState (value: JobState) {
-  return exists(value) && jobStates.indexOf(value) !== -1
+  return exists(value) && jobStates.includes(value)
+}
+
+function isValidJobType (value: any) {
+  return exists(value) && jobTypes.includes(value)
 }
 
 // ---------------------------------------------------------------------------
 
 export {
-  isValidJobState
+  isValidJobState,
+  isValidJobType
 }
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 3c810da98..ec601e9ea 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -121,11 +121,20 @@ class JobQueue {
     return queue.add(obj.payload, jobArgs)
   }
 
-  async listForApi (state: JobState, start: number, count: number, asc?: boolean): Promise<Bull.Job[]> {
+  async listForApi (options: {
+    state: JobState,
+    start: number,
+    count: number,
+    asc?: boolean,
+    jobType: JobType
+  }): Promise<Bull.Job[]> {
+    const { state, start, count, asc, jobType } = options
     let results: Bull.Job[] = []
 
+    const filteredJobTypes = this.filterJobTypes(jobType)
+
     // TODO: optimize
-    for (const jobType of jobTypes) {
+    for (const jobType of filteredJobTypes) {
       const queue = this.queues[ jobType ]
       if (queue === undefined) {
         logger.error('Unknown queue %s to list jobs.', jobType)
@@ -149,10 +158,12 @@ class JobQueue {
     return results.slice(start, start + count)
   }
 
-  async count (state: JobState): Promise<number> {
+  async count (state: JobState, jobType?: JobType): Promise<number> {
     let total = 0
 
-    for (const type of jobTypes) {
+    const filteredJobTypes = this.filterJobTypes(jobType)
+
+    for (const type of filteredJobTypes) {
       const queue = this.queues[ type ]
       if (queue === undefined) {
         logger.error('Unknown queue %s to count jobs.', type)
@@ -180,6 +191,12 @@ class JobQueue {
     })
   }
 
+  private filterJobTypes (jobType?: JobType) {
+    if (!jobType) return jobTypes
+
+    return jobTypes.filter(t => t === jobType)
+  }
+
   static get Instance () {
     return this.instance || (this.instance = new this())
   }
@@ -188,5 +205,6 @@ class JobQueue {
 // ---------------------------------------------------------------------------
 
 export {
+  jobTypes,
   JobQueue
 }
diff --git a/server/middlewares/validators/jobs.ts b/server/middlewares/validators/jobs.ts
index 41a8d6899..b57615dbc 100644
--- a/server/middlewares/validators/jobs.ts
+++ b/server/middlewares/validators/jobs.ts
@@ -1,13 +1,17 @@
 import * as express from 'express'
-import { param } from 'express-validator'
-import { isValidJobState } from '../../helpers/custom-validators/jobs'
+import { param, query } from 'express-validator'
+import { isValidJobState, isValidJobType } from '../../helpers/custom-validators/jobs'
 import { logger } from '../../helpers/logger'
 import { areValidationErrors } from './utils'
 
 const listJobsValidator = [
-  param('state').custom(isValidJobState).not().isEmpty().withMessage('Should have a valid job state'),
+  param('state')
+    .custom(isValidJobState).not().isEmpty().withMessage('Should have a valid job state'),
+  query('jobType')
+    .optional()
+    .custom(isValidJobType).withMessage('Should have a valid job state'),
 
-  async (req: express.Request, res: express.Response, next: express.NextFunction) => {
+  (req: express.Request, res: express.Response, next: express.NextFunction) => {
     logger.debug('Checking listJobsValidator parameters.', { parameters: req.params })
 
     if (areValidationErrors(req, res)) return
diff --git a/server/tests/api/check-params/jobs.ts b/server/tests/api/check-params/jobs.ts
index c70139514..22e237964 100644
--- a/server/tests/api/check-params/jobs.ts
+++ b/server/tests/api/check-params/jobs.ts
@@ -51,6 +51,17 @@ describe('Test jobs API validators', function () {
       })
     })
 
+    it('Should fail with an incorrect job type', async function () {
+      await makeGetRequest({
+        url: server.url,
+        token: server.accessToken,
+        path,
+        query: {
+          jobType: 'toto'
+        }
+      })
+    })
+
     it('Should fail with a bad start pagination', async function () {
       await checkBadStartPagination(server.url, path, server.accessToken)
     })
@@ -79,6 +90,7 @@ describe('Test jobs API validators', function () {
         statusCodeExpected: 403
       })
     })
+
   })
 
   after(async function () {
diff --git a/server/tests/api/server/handle-down.ts b/server/tests/api/server/handle-down.ts
index a0f505474..7e36067f1 100644
--- a/server/tests/api/server/handle-down.ts
+++ b/server/tests/api/server/handle-down.ts
@@ -184,7 +184,14 @@ describe('Test handle downs', function () {
     const states: JobState[] = [ 'waiting', 'active' ]
 
     for (const state of states) {
-      const res = await getJobsListPaginationAndSort(servers[ 0 ].url, servers[ 0 ].accessToken, state,0, 50, '-createdAt')
+      const res = await getJobsListPaginationAndSort({
+        url: servers[ 0 ].url,
+        accessToken: servers[ 0 ].accessToken,
+        state: state,
+        start: 0,
+        count: 50,
+        sort: '-createdAt'
+      })
       expect(res.body.data).to.have.length(0)
     }
   })
diff --git a/server/tests/api/server/jobs.ts b/server/tests/api/server/jobs.ts
index ceea47a85..58d8c8c10 100644
--- a/server/tests/api/server/jobs.ts
+++ b/server/tests/api/server/jobs.ts
@@ -41,20 +41,46 @@ describe('Test jobs', function () {
     expect(res.body.data).to.have.length.above(2)
   })
 
-  it('Should list jobs with sort and pagination', async function () {
-    const res = await getJobsListPaginationAndSort(servers[1].url, servers[1].accessToken, 'completed', 1, 2, 'createdAt')
-    expect(res.body.total).to.be.above(2)
-    expect(res.body.data).to.have.lengthOf(2)
+  it('Should list jobs with sort, pagination and job type', async function () {
+    {
+      const res = await getJobsListPaginationAndSort({
+        url: servers[ 1 ].url,
+        accessToken: servers[ 1 ].accessToken,
+        state: 'completed',
+        start: 1,
+        count: 2,
+        sort: 'createdAt'
+      })
+      expect(res.body.total).to.be.above(2)
+      expect(res.body.data).to.have.lengthOf(2)
+
+      let job: Job = res.body.data[ 0 ]
+      // Skip repeat jobs
+      if (job.type === 'videos-views') job = res.body.data[ 1 ]
+
+      expect(job.state).to.equal('completed')
+      expect(job.type.startsWith('activitypub-')).to.be.true
+      expect(dateIsValid(job.createdAt as string)).to.be.true
+      expect(dateIsValid(job.processedOn as string)).to.be.true
+      expect(dateIsValid(job.finishedOn as string)).to.be.true
+    }
 
-    let job = res.body.data[0]
-    // Skip repeat jobs
-    if (job.type === 'videos-views') job = res.body.data[1]
+    {
+      const res = await getJobsListPaginationAndSort({
+        url: servers[ 1 ].url,
+        accessToken: servers[ 1 ].accessToken,
+        state: 'completed',
+        start: 0,
+        count: 100,
+        sort: 'createdAt',
+        jobType: 'activitypub-http-broadcast'
+      })
+      expect(res.body.total).to.be.above(2)
 
-    expect(job.state).to.equal('completed')
-    expect(job.type.startsWith('activitypub-')).to.be.true
-    expect(dateIsValid(job.createdAt)).to.be.true
-    expect(dateIsValid(job.processedOn)).to.be.true
-    expect(dateIsValid(job.finishedOn)).to.be.true
+      for (const j of res.body.data as Job[]) {
+        expect(j.type).to.equal('activitypub-http-broadcast')
+      }
+    }
   })
 
   after(async function () {
diff --git a/server/tests/real-world/real-world.ts b/server/tests/real-world/real-world.ts
index 8b070004d..cba5ac311 100644
--- a/server/tests/real-world/real-world.ts
+++ b/server/tests/real-world/real-world.ts
@@ -354,7 +354,14 @@ async function isTherePendingRequests (servers: ServerInfo[]) {
   // Check if each server has pending request
   for (const server of servers) {
     for (const state of states) {
-      const p = getJobsListPaginationAndSort(server.url, server.accessToken, state, 0, 10, '-createdAt')
+      const p = getJobsListPaginationAndSort({
+        url: server.url,
+        accessToken: server.accessToken,
+        state: state,
+        start: 0,
+        count: 10,
+        sort: '-createdAt'
+      })
         .then(res => {
           if (res.body.total > 0) pendingRequests = true
         })