feat: moving args to run function

Author: merencia

migrations/postgres/2_create_jobs_table.js

@@ -8,6 +8,7 @@ exports.up = async function(knex) {
     table.string('class').notNullable();
     table.jsonb('args').notNullable();
 
+    table.integer('timeout').nullable();
     table.integer('attempt').notNullable().defaultTo(0);
     table.integer('max_attempts').notNullable().defaultTo(5);
     table.specificType('result', 'jsonb').notNullable().defaultTo('{}');

src/backends/postgres/postgres-backend.ts

@@ -43,7 +43,8 @@ export class PostgresBackend implements Backend{
       queue: job.queue,
       class: job.class,
       script: job.script,
-      args: this.knex.raw('?', [JSON.stringify(job.args)])
+      args: this.knex.raw('?', [JSON.stringify(job.args)]),
+      timeout: job.timeout,
     }
 
     const inserted = await this.knex('sidequest_jobs').insert(data).returning('*');
@@ -84,6 +85,7 @@ export class PostgresBackend implements Backend{
       state: job.state,
       script: job.script,
       class: job.class,
+      timeout: job.timeout,
       attempt: job.attempt,
       max_attempts: job.max_attempts,
       errors: job.errors,

src/core/job.ts

@@ -23,20 +23,14 @@ export abstract class Job {
     this.timeout = options.timeout;
   }
 
-  abstract run(): any | Promise<any>;
+  abstract run(...args: any[]): any | Promise<any>;
+
+  static config(this: { new (...args: any[]): Job }, jobOptions: JobOptions){
+    return new JobBuilder(this).config(jobOptions);
+  }
 
   static enqueue(this: { new (...args: any[]): Job }, ...args: any[]): JobData | Promise<JobData> {
-    const job = new this(...args);
-    const backend = Sidequest.getBackend();
-    const jobData: JobData = {
-      queue: job.queue,
-      script: job.script,
-      class: job.class,
-      args: args,
-      attempt: 0,
-      max_attempts: 5
-    }
-    return backend.insertJob(jobData);
+    return new JobBuilder(this).enqueue(args);
   }
 }
 
@@ -52,4 +46,36 @@ function buildPath() {
   }
 
   throw new Error('Could not determine the task path');
+}
+
+class JobBuilder{
+  JobClass: new (...args: any[]) => Job;
+  job?: Job;
+
+  constructor(JobClass: { new (...args: any[]): Job }){
+    this.JobClass = JobClass;
+  }
+
+  config(options: JobOptions){
+    this.job = new this.JobClass(options);
+    return this;
+  }
+
+  enqueue(...args: any[]){
+    if(!this.job){
+      this.job = new this.JobClass({ queue: 'default' });
+    }
+
+    const backend = Sidequest.getBackend();
+    const jobData: JobData = {
+      queue: this.job.queue,
+      script: this.job.script,
+      class: this.job.class,
+      args: args,
+      attempt: 0,
+      max_attempts: 5,
+      timeout: this.job.timeout
+    }
+    return backend.insertJob(jobData);
+  }
 }

src/core/schema/job-data.ts

@@ -7,6 +7,7 @@ export type JobData = {
     script: string;
     class: string;
     args: any[];
+    timeout?: number;
     attempt: number;
     max_attempts: number;
     result?: any;

src/test-jobs/dynamic-dummy-job.ts

@@ -1,12 +1,6 @@
 import { Job } from '../sidequest';
 
 export class DynamicDummyJob extends Job {
-  constructor(queue: string){
-    super({
-      queue: queue
-    })
-  }
-
   async run(): Promise<any> {
     await new Promise((r) => { setTimeout(r, 800)})
     return 'dummy job';

src/workers/executor.test.ts

@@ -17,7 +17,7 @@ describe('executror.ts', ()=>{
         run: async () => 'ok'
       };
 
-      const result = await executeTask(job as any);
+      const result = await executeTask(job as any, []);
       assert.equal(result, 'ok');
     });
 
@@ -27,7 +27,7 @@ describe('executror.ts', ()=>{
       };
 
       try {
-        await executeTask(job as any);
+        await executeTask(job as any, []);
         assert.fail('Expected error, but promise resolved');
       } catch (err: any) {
         assert.instanceOf(err, Error);
@@ -43,7 +43,7 @@ describe('executror.ts', ()=>{
       };
 
       try {
-        await executeTask(job as any);
+        await executeTask(job as any, []);
         assert.fail('Expected timeout, but promise resolved');
       } catch (err: any) {
         assert.instanceOf(err, Error);

src/workers/executor.ts

@@ -15,12 +15,12 @@ export async function execute(jobData: JobData, config: SidequestConfig): Promis
   }
 
   const args = jobData.args;
-  const job: Job = new JobClass(...args);
+  const job: Job = new JobClass({ queue: jobData.queue, timeout: jobData.timeout });
 
   jobData = await JobActions.setRunning(jobData);
   try {
     logger().info(`Running job ${jobData.class} with args: ${JSON.stringify(jobData.args)}`);
-    const result = await executeTask(job);
+    const result = await executeTask(job, args);
     jobData = await JobActions.setComplete(jobData, result);
     logger().info(`Job ${jobData.class} has completed with args: ${JSON.stringify(jobData.args)}`);
   } catch (error: any){
@@ -29,9 +29,10 @@ export async function execute(jobData: JobData, config: SidequestConfig): Promis
   }
 }
 
-export function executeTask(job: Job){
+export function executeTask(job: Job, args: any[]){
   const promises: Promise<any>[] = [];
 
+  console.log(job)
   if(job.timeout){
     const timeout = new Promise((resolve, reject) => setTimeout(()=> {
       reject(new Error(`Job ${job.class} timed out: ${ JSON.stringify(job) }`));
@@ -41,7 +42,7 @@ export function executeTask(job: Job){
 
   const run = new Promise(async (resolve, reject) => {
     try {
-      const result = await job.run();
+      const result = await job.run(...args);
       resolve(result);
     } catch(error: any){
       reject(error);

src/workers/main.test.ts

@@ -77,9 +77,9 @@ describe('main.ts', ()=> {
 
     await worker.run(config);
 
-    await DynamicDummyJob.enqueue(lowQueueName);
-    await DynamicDummyJob.enqueue(mediumQueueName);
-    await DynamicDummyJob.enqueue(highQueueName);
+    await DynamicDummyJob.config({queue: lowQueueName }).enqueue();
+    await DynamicDummyJob.config({queue: mediumQueueName }).enqueue();
+    await DynamicDummyJob.config({queue: highQueueName }).enqueue();
 
     const executed: JobData[] = [];