shared/extra-utils/server/jobs.ts

   1 import * as request from 'supertest'
   2 import { HttpStatusCode } from '../../../shared/core-utils/miscs/http-error-codes'
   3 import { getDebug, makeGetRequest } from '../../../shared/extra-utils'
   4 import { Job, JobState, JobType, ServerDebug } from '../../models'
   5 import { wait } from '../miscs/miscs'
   6 import { ServerInfo } from './servers'
   7
   8 function buildJobsUrl (state?: JobState) {
   9   let path = '/api/v1/jobs'
  10
  11   if (state) path += '/' + state
  12
  13   return path
  14 }
  15
  16 function getJobsList (url: string, accessToken: string, state?: JobState) {
  17   const path = buildJobsUrl(state)
  18
  19   return request(url)
  20     .get(path)
  21     .set('Accept', 'application/json')
  22     .set('Authorization', 'Bearer ' + accessToken)
  23     .expect(HttpStatusCode.OK_200)
  24     .expect('Content-Type', /json/)
  25 }
  26
  27 function getJobsListPaginationAndSort (options: {
  28   url: string
  29   accessToken: string
  30   start: number
  31   count: number
  32   sort: string
  33   state?: JobState
  34   jobType?: JobType
  35 }) {
  36   const { url, accessToken, state, start, count, sort, jobType } = options
  37   const path = buildJobsUrl(state)
  38
  39   const query = {
  40     start,
  41     count,
  42     sort,
  43     jobType
  44   }
  45
  46   return makeGetRequest({
  47     url,
  48     path,
  49     token: accessToken,
  50     statusCodeExpected: HttpStatusCode.OK_200,
  51     query
  52   })
  53 }
  54
  55 async function waitJobs (serversArg: ServerInfo[] | ServerInfo) {
  56   const pendingJobWait = process.env.NODE_PENDING_JOB_WAIT
  57     ? parseInt(process.env.NODE_PENDING_JOB_WAIT, 10)
  58     : 250
  59
  60   let servers: ServerInfo[]
  61
  62   if (Array.isArray(serversArg) === false) servers = [ serversArg as ServerInfo ]
  63   else servers = serversArg as ServerInfo[]
  64
  65   const states: JobState[] = [ 'waiting', 'active', 'delayed' ]
  66   const repeatableJobs = [ 'videos-views', 'activitypub-cleaner' ]
  67   let pendingRequests: boolean
  68
  69   function tasksBuilder () {
  70     const tasks: Promise<any>[] = []
  71
  72     // Check if each server has pending request
  73     for (const server of servers) {
  74       for (const state of states) {
  75         const p = getJobsListPaginationAndSort({
  76           url: server.url,
  77           accessToken: server.accessToken,
  78           state: state,
  79           start: 0,
  80           count: 10,
  81           sort: '-createdAt'
  82         }).then(res => res.body.data)
  83           .then((jobs: Job[]) => jobs.filter(j => !repeatableJobs.includes(j.type)))
  84           .then(jobs => {
  85             if (jobs.length !== 0) {
  86               pendingRequests = true
  87             }
  88           })
  89
  90         tasks.push(p)
  91       }
  92
  93       const p = getDebug(server.url, server.accessToken)
  94         .then(res => res.body)
  95         .then((obj: ServerDebug) => {
  96           if (obj.activityPubMessagesWaiting !== 0) {
  97             pendingRequests = true
  98           }
  99         })
 100
 101       tasks.push(p)
 102     }
 103
 104     return tasks
 105   }
 106
 107   do {
 108     pendingRequests = false
 109     await Promise.all(tasksBuilder())
 110
 111     // Retry, in case of new jobs were created
 112     if (pendingRequests === false) {
 113       await wait(pendingJobWait)
 114       await Promise.all(tasksBuilder())
 115     }
 116
 117     if (pendingRequests) {
 118       await wait(pendingJobWait)
 119     }
 120   } while (pendingRequests)
 121 }
 122
 123 // ---------------------------------------------------------------------------
 124
 125 export {
 126   getJobsList,
 127   waitJobs,
 128   getJobsListPaginationAndSort
 129 }