3 files changed, 27 insertions, 14 deletions

diff --git a/server/lib/jobs/handlers/index.ts b/server/lib/jobs/handlers/index.ts
index ae5440031..7d0263b15 100644
--- a/server/lib/jobs/handlers/index.ts
+++ b/server/lib/jobs/handlers/index.ts
@@ -1,6 +1,14 @@
 import * as videoTranscoder from './video-transcoder'
 
-const jobHandlers = {
+import { VideoInstance } from '../../../models'
+
+export interface JobHandler<T> {
+  process (data: object, callback: (err: Error, videoInstance?: T) => void)
+  onError (err: Error, jobId: number, video: T, callback: (err: Error) => void)
+  onSuccess (data: any, jobId: number, video: T, callback: (err: Error) => void)
+}
+
+const jobHandlers: { [ handlerName: string ]: JobHandler<any> } = {
   videoTranscoder
 }
 
diff --git a/server/lib/jobs/handlers/video-transcoder.ts b/server/lib/jobs/handlers/video-transcoder.ts
index 43599356a..efa18ef2d 100644
--- a/server/lib/jobs/handlers/video-transcoder.ts
+++ b/server/lib/jobs/handlers/video-transcoder.ts
@@ -1,8 +1,9 @@
 import { database as db } from '../../../initializers/database'
 import { logger } from '../../../helpers'
 import { addVideoToFriends } from '../../../lib'
+import { VideoInstance } from '../../../models'
 
-function process (data, callback) {
+function process (data: { id: string }, callback: (err: Error, videoInstance?: VideoInstance) => void) {
   db.Video.loadAndPopulateAuthorAndPodAndTags(data.id, function (err, video) {
     if (err) return callback(err)
 
@@ -12,12 +13,12 @@ function process (data, callback) {
   })
 }
 
-function onError (err, jobId, video, callback) {
+function onError (err: Error, jobId: number, video: VideoInstance, callback: () => void) {
   logger.error('Error when transcoding video file in job %d.', jobId, { error: err })
   return callback()
 }
 
-function onSuccess (data, jobId, video, callback) {
+function onSuccess (data: any, jobId: number, video: VideoInstance, callback: (err: Error) => void) {
   logger.info('Job %d is a success.', jobId)
 
   video.toAddRemoteJSON(function (err, remoteVideo) {
diff --git a/server/lib/jobs/job-scheduler.ts b/server/lib/jobs/job-scheduler.ts
index ad5f7f6d9..2f01387e7 100644
--- a/server/lib/jobs/job-scheduler.ts
+++ b/server/lib/jobs/job-scheduler.ts
@@ -1,4 +1,5 @@
 import { forever, queue } from 'async'
+import * as Sequelize from 'sequelize'
 
 import { database as db } from '../../initializers/database'
 import {
@@ -7,7 +8,10 @@ import {
   JOB_STATES
 } from '../../initializers'
 import { logger } from '../../helpers'
-import { jobHandlers } from './handlers'
+import { JobInstance } from '../../models'
+import { JobHandler, jobHandlers } from './handlers'
+
+type JobQueueCallback = (err: Error) => void
 
 class JobScheduler {
 
@@ -24,7 +28,7 @@ class JobScheduler {
 
     logger.info('Jobs scheduler activated.')
 
-    const jobsQueue = queue(this.processJob.bind(this))
+    const jobsQueue = queue<JobInstance, JobQueueCallback>(this.processJob.bind(this))
 
     // Finish processing jobs from a previous start
     const state = JOB_STATES.PROCESSING
@@ -58,7 +62,7 @@ class JobScheduler {
     })
   }
 
-  createJob (transaction, handlerName: string, handlerInputData: object, callback) {
+  createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object, callback: (err: Error) => void) {
     const createQuery = {
       state: JOB_STATES.PENDING,
       handlerName,
@@ -69,7 +73,7 @@ class JobScheduler {
     db.Job.create(createQuery, options).asCallback(callback)
   }
 
-  private enqueueJobs (err, jobsQueue, jobs) {
+  private enqueueJobs (err: Error, jobsQueue: AsyncQueue<JobInstance>, jobs: JobInstance[]) {
     if (err) {
       logger.error('Cannot list pending jobs.', { error: err })
     } else {
@@ -79,7 +83,7 @@ class JobScheduler {
     }
   }
 
-  private processJob (job, callback) {
+  private processJob (job: JobInstance, callback: (err: Error) => void) {
     const jobHandler = jobHandlers[job.handlerName]
 
     logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
@@ -89,8 +93,8 @@ class JobScheduler {
       if (err) return this.cannotSaveJobError(err, callback)
 
       if (jobHandler === undefined) {
-        logger.error('Unknown job handler for job %s.', jobHandler.handlerName)
-        return callback()
+        logger.error('Unknown job handler for job %s.', job.handlerName)
+        return callback(null)
       }
 
       return jobHandler.process(job.handlerInputData, (err, result) => {
@@ -104,7 +108,7 @@ class JobScheduler {
     })
   }
 
-  private onJobError (jobHandler, job, jobResult, callback) {
+  private onJobError (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any, callback: (err: Error) => void) {
     job.state = JOB_STATES.ERROR
 
     job.save().asCallback(err => {
@@ -114,7 +118,7 @@ class JobScheduler {
     })
   }
 
-  private onJobSuccess (jobHandler, job, jobResult, callback) {
+  private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any, callback: (err: Error) => void) {
     job.state = JOB_STATES.SUCCESS
 
     job.save().asCallback(err => {
@@ -124,7 +128,7 @@ class JobScheduler {
     })
   }
 
-  private cannotSaveJobError (err, callback) {
+  private cannotSaveJobError (err: Error, callback: (err: Error) => void) {
     logger.error('Cannot save new job state.', { error: err })
     return callback(err)
   }