Jobs

Jobs in Orionjs enable you to execute background tasks, schedule recurring operations, and handle asynchronous processing. Powered by the @orion-js/dogs package, jobs provide a robust way to manage both recurring and event-driven background operations.

Installation

pnpm add @orion-js/dogs

Defining Jobs Controllers

The recommended way to define jobs in Orionjs is to use the @Jobs() decorator on a class and the job-specific decorators on methods:

import {Jobs, RecurrentJob, EventJob, createEventJob, createRecurrentJob} from '@orion-js/dogs'
import {logger} from '@orion-js/logger'
import {Inject} from '@orion-js/services'
import {ExampleService} from '../services/ExampleService'

@Jobs()
export default class ExampleJobs {
  @Inject(() => ExampleService)
  private exampleService: ExampleService

  // Define a recurrent job using the method decorator
  @RecurrentJob()
  createExamples = createRecurrentJob({
    runEvery: '1m', // Run every minute
    resolve: async () => {
      logger.info('Creating example...')
      await this.exampleService.makeExample()
    }
  })

  // Define an event job using the method decorator
  @EventJob()
  sendWelcomeEmail = createEventJob({
    params: {
      email: { type: String },
      name: { type: String }
    },
    resolve: async (params) => {
      const {email, name} = params
      logger.info('Sending welcome email...', {email})
      await this.exampleService.sendWelcomeEmail(email, name)
    }
  })

  // Regular method to schedule an event job
  async registerUser(userData) {
    // Create user account
    const user = await this.exampleService.createUser(userData)

    // Schedule welcome email job
    await this.sendWelcomeEmail.schedule({
      params: {
        email: user.email,
        name: user.name
      }
    })

    return user
  }
}

With this approach:

The class is decorated with @Jobs() to mark it as a job controller
Each job is defined as a class property using createRecurrentJob or createEventJob with its specific decorator (@RecurrentJob or @EventJob)
Job parameters are validated based on the schema provided in the params option
You can use dependency injection with @Inject(() => Service)
The class can also include regular methods that aren’t jobs

Types of Jobs

Orionjs supports two types of jobs:

1. Recurrent Jobs

Jobs that run at specified intervals or on a schedule:

@Jobs()
export default class MaintenanceJobs {
  @Inject(() => DatabaseService)
  private databaseService: DatabaseService

  @RecurrentJob()
  dailyCleanup = createRecurrentJob({
    // Run every 24 hours
    runEvery: '24h',
    // Job priority (higher is more important, default: 100)
    priority: 100,
    // Optional: specify error handling
    onError: async function(error, params, context) {
      logger.error('Daily cleanup failed', {error})
      if (context.tries < 3) {
        return {action: 'retry', runIn: 60000} // Retry in 1 minute
      }
      return {action: 'dismiss'}
    },
    resolve: async () => {
      // Your job logic here
      await this.databaseService.cleanupOldRecords()
      return {status: 'success'}
    }
  })

  // Alternative: use getNextRun instead of runEvery
  @RecurrentJob()
  nightlyBackup = createRecurrentJob({
    getNextRun: () => {
      const nextRun = new Date()
      nextRun.setHours(3, 0, 0, 0) // Run at 3:00 AM
      if (nextRun < new Date()) {
        nextRun.setDate(nextRun.getDate() + 1) // Tomorrow if already past 3 AM
      }
      return nextRun
    },
    resolve: async () => {
      // Backup logic
    }
  })
}

2. Event Jobs

Jobs that run on-demand when triggered by an event:

@Jobs()
export default class NotificationJobs {
  @Inject(() => EmailService)
  private emailService: EmailService

  @EventJob()
  sendEmail = createEventJob({
    params: {
      to: { type: String },
      subject: { type: String },
      body: { type: String }
    },
    // Optional: limit how many sendEmail jobs can run in parallel on this server
    maxParallelExecutionsPerServer: 2,
    // Optional: configure error handling
    onError: async function(error, params, context) {
      if (context.tries < 3) {
        return {
          action: 'retry',
          runIn: Math.pow(2, context.tries) * 1000 // Exponential backoff
        }
      }
      return {action: 'dismiss'}
    },
    // Optional: configure stale job handling
    onStale: async function(params, context) {
      logger.warn('Email send job became stale', {params})
    },
    resolve: async (params) => {
      const {to, subject, body} = params
      
      // Use injected services
      await this.emailService.send(to, subject, body)
      
      return {sent: true}
    }
  })
}

Job Creation Options

RecurrentJob Options

Option	Type	Description
`runEvery`	`number \| string`	Milliseconds between runs or a string like ‘1d’, ‘4h’, etc. Ignored if `getNextRun` is provided.
`getNextRun`	`Function`	Function returning a Date for the next execution. Takes precedence over `runEvery`.
`priority`	`number`	Job priority. Higher numbers get executed first (default: 100).
`lockTime`	`number`	Override the global `defaultLockTime` for this job only.
`onError`	`Function`	Called if the job fails. Can be used to retry or dismiss the job.
`onStale`	`Function`	Called if the job locktime is expired.
`saveExecutionsFor`	`number`	Time in milliseconds to keep job execution history (default: 1 week). Set to 0 to disable.
`resolve`	`Function`	The function that contains the job’s logic.

EventJob Options

Option	Type	Description
`params`	`Object \| Schema`	Schema for validating job parameters.
`lockTime`	`number`	Override the global `defaultLockTime` for this job only.
`maxTries`	`number`	Override the global event `maxTries` for this job only.
`maxParallelExecutionsPerServer`	`number`	Limit how many executions of this job can run at the same time in a single server process.
`onError`	`Function`	Called if the job fails. Can be used to retry or dismiss the job.
`onStale`	`Function`	Called if the job locktime is expired.
`saveExecutionsFor`	`number`	Time in milliseconds to keep job execution history (default: 1 week). Set to 0 to disable.
`resolve`	`Function`	The function that contains the job’s logic and receives validated params.

Starting Workers

In your application setup (typically in your main file), you start workers to process jobs:

import {startWorkers} from '@orion-js/dogs'

// One scheduler discovers jobs defined using the @Jobs() decorator and runs up to four at once.
const workersInstance = startWorkers({
  workersCount: 4, // Maximum concurrent executions (default: 4)
  nPartitions: 16, // MongoDB acquisition partitions (default: 1)
  pollInterval: 3000, // Milliseconds between polling for new jobs (default: 3000)
  defaultLockTime: 30000, // Default lock time in milliseconds (default: 30000 = 30 seconds)
  maxTries: 10, // Optional. Applies only to event jobs
  maxTriesReachedRetentionMs: 7 * 24 * 60 * 60 * 1000, // Default: 1 week. Use null to keep forever
  onMaxTriesReached: async (job) => {
    logger.error('Job reached max tries', {jobName: job.name, jobId: job.jobId})
  }
})

// To stop workers (e.g., when shutting down the application)
process.on('SIGTERM', async () => {
  await workersInstance.stop()
  process.exit(0)
})

Each startWorkers() call creates one scheduler loop, regardless of workersCount. The scheduler claims jobs atomically until its concurrency is full. It queries one partition at a time, advances through a shuffled order, and waits pollInterval only after a complete empty sweep. This avoids a $in across partitions and keeps each acquisition on an equality-prefixed index. Priority is strict inside each partition and approximate across partitions. Use the same nPartitions value in every application instance. Dogs stores no worker registry or partition metadata. A background reconciler gradually assigns legacy records and repairs partitions outside the configured range without moving actively locked jobs. workersInstance.runningExecutions reports how many active executions currently consume scheduler capacity. maxTriesReachedRetentionMs controls how long terminal event-job records remain in MongoDB. It defaults to one week; set it to null to retain them indefinitely. Dogs stores an expiresAt date, ensures a TTL index for that field, and applies the configured retention to existing maxTriesReached records whenever workers start. Use maxParallelExecutionsPerServer when a job is safe to queue but should not run too many times concurrently on the same server, for example:

@EventJob()
generateInvoicePdf = createEventJob({
  maxParallelExecutionsPerServer: 1,
  params: {
    invoiceId: {type: String}
  },
  resolve: async ({invoiceId}) => {
    // Avoid saturating CPU or a shared local resource on this server
    await generatePdf(invoiceId)
  }
})

This limit only applies to event jobs. Recurrent jobs already execute as a single job record, so they do not need this option.

Scheduling Event Jobs

Trigger an event job to run:

// Assuming you have this job defined
@EventJob()
sendEmail = createEventJob({
  params: {
    to: { type: String },
    subject: { type: String },
    body: { type: String }
  },
  resolve: async (params) => {
    // Job logic
  }
})

// Run immediately
await this.sendEmail.schedule({
  params: {
    to: 'user@example.com',
    subject: 'Welcome!',
    body: 'Welcome to our platform.'
  }
})

// Run after 5 minutes
await this.sendEmail.schedule({
  params: {
    to: 'user@example.com',
    subject: 'Reminder',
    body: 'Don\'t forget to complete your profile.'
  },
  runIn: 5 * 60 * 1000 // 5 minutes in milliseconds
})

// Run at a specific time
await this.sendEmail.schedule({
  params: {
    to: 'user@example.com',
    subject: 'Scheduled Announcement',
    body: 'New features are now available!'
  },
  runAt: new Date('2023-12-31T00:00:00')
})

// Ensure uniqueness
await this.sendEmail.schedule({
  params: {
    to: 'user@example.com',
    subject: 'Daily Update',
    body: 'Daily update content'
  },
  uniqueIdentifier: 'daily-update-user123', // Won't schedule duplicates with this identifier
  replace: true // Replace existing job with this identifier if it exists
})

Job Execution Context

The job’s resolve function receives additional parameters:

@RecurrentJob()
exampleJob = createRecurrentJob({
  runEvery: '1h',
  resolve: async (_, context) => {
    // context contains information about the job execution
    console.log('Job ID:', context.jobId)
    console.log('Execution attempt:', context.tries)
    console.log('Scheduled at:', context.createdAt)
    
    // Your job logic here
  }
})

Error Handling

Jobs support sophisticated error handling:

@EventJob()
riskyJob = createEventJob({
  params: {
    userId: { type: String }
  },
  onError: async (error, params, context) => {
    logger.error('Job failed', {
      error,
      userId: params.userId,
      attempt: context.tries
    })
    
    if (context.tries < 3) {
      // Retry with exponential backoff
      return {
        action: 'retry',
        runIn: Math.pow(2, context.tries) * 1000
      }
    }
    
    // After 3 failed attempts, send an alert and dismiss the job
    await sendAlertToAdmin(`Job failed for user ${params.userId}: ${error.message}`)
    return { action: 'dismiss' }
  },
  resolve: async (params) => {
    // Job implementation
  }
})

Monitoring Jobs

You can monitor job status and history:

import {getJobStatus, getJobHistory} from '@orion-js/dogs'

// Get status of all jobs
const status = await getJobStatus()
console.log('Active jobs:', status.active)
console.log('Scheduled jobs:', status.scheduled)

// Get history of a specific job
const history = await getJobHistory('sendEmail')
console.log('Recent executions:', history.executions)
console.log('Success rate:', history.stats.successRate)

Best Practices

Keep Jobs Idempotent: Design jobs so they can be safely re-executed without side effects.
Use Parameters Validation: Define schemas for job parameters to ensure they’re valid.
Handle Errors Properly: Implement robust error handling with appropriate retry strategies.
Set Realistic Lock Times: Ensure the lockTime is longer than a job’s expected execution time.
Monitor Job Performance: Keep track of job durations to optimize processing.
Organize by Domain: Group related jobs in domain-specific controller classes.
Test Jobs Thoroughly: Write unit tests for job logic to ensure reliability.
Check Job Status: Implement health checks to monitor job processing.

Get Started

Essentials

Controllers

Other Modules

Installation

Defining Jobs Controllers

Types of Jobs

1. Recurrent Jobs

2. Event Jobs

Job Creation Options

RecurrentJob Options

EventJob Options

Starting Workers

Scheduling Event Jobs

Job Execution Context

Error Handling

Monitoring Jobs

Best Practices

​Installation

​Defining Jobs Controllers

​Types of Jobs

​1. Recurrent Jobs

​2. Event Jobs

​Job Creation Options

​RecurrentJob Options

​EventJob Options

​Starting Workers

​Scheduling Event Jobs

​Job Execution Context

​Error Handling

​Monitoring Jobs

​Best Practices

Installation

Defining Jobs Controllers

Types of Jobs

1. Recurrent Jobs

2. Event Jobs

Job Creation Options

RecurrentJob Options

EventJob Options

Starting Workers

Scheduling Event Jobs

Job Execution Context

Error Handling

Monitoring Jobs

Best Practices