Quick Start

Build your first job queue in 5 minutes with Platformatic Job Queue

Quick Start

This guide walks you through creating a complete working job queue from scratch. You’ll learn how to create a queue, register a handler, enqueue jobs, and handle results.

Prerequisites

• Node.js 22.19.0 or later installed
• @platformatic/job-queue package installed (Installation Guide)

Complete Example

Let’s build an email processing queue that demonstrates all core features:

1. Create the queue with storage

   First, import the required classes and create a queue with in-memory storage:

   index.ts

   import { Queue, MemoryStorage } from '@platformatic/job-queue'
   
   // Create a queue with in-memory storage
   const storage = new MemoryStorage()
   const queue = new Queue<{ email: string }, { sent: boolean }>({
     storage,
     concurrency: 5
   })
   

   The generic types <TPayload, TResult> provide full type safety for your job data and results.

2. Register a job handler

   Define how your queue should process jobs:

   index.ts

   // Register a job handler
   queue.execute(async (job) => {
     console.log(`Processing job ${job.id}:`, job.payload)
     // Send email...
     return { sent: true }
   })
   

   The handler receives a Job object with:

   • id - Unique job identifier
   • payload - The job data (typed as { email: string })
   • attempts - Current attempt number (starts at 1)
   • signal - AbortSignal for cancellation
3. Start the queue

   Connect to storage and start processing:

   index.ts

   // Start the queue
   await queue.start()
   console.log('Queue started and ready to process jobs')
   
4. Enqueue jobs (fire-and-forget)

   Add jobs to the queue without waiting for results:

   index.ts

   // Enqueue jobs
   await queue.enqueue('email-1', { email: '[e-mail]' })
   console.log('Job enqueued')
   
   // Optional per-job TTL override for cached result/error
   await queue.enqueue('email-ttl', { email: '[e-mail]' }, {
     resultTTL: 5 * 60 * 1000 // 5 minutes
   })
   

   Jobs are deduplicated by ID. If you enqueue the same ID twice, it will only be processed once.

5. Enqueue and wait for results

   For request/response patterns, use enqueueAndWait():

   index.ts

   // Or wait for the result
   const result = await queue.enqueueAndWait('email-2', { email: '[e-mail]' }, {
     timeout: 30000,
     resultTTL: 24 * 60 * 60 * 1000 // keep this result for 24h
   })
   console.log('Result:', result) // { sent: true }
   

   This will:

   1. Enqueue the job
   2. Wait for it to be processed
   3. Return the result (or throw if it fails)
6. Graceful shutdown

   Stop the queue gracefully, waiting for in-flight jobs to complete:

   index.ts

   // Graceful shutdown
   await queue.stop()
   console.log('Queue stopped gracefully')
   

Complete Working Code

Here’s the full example in one file:

index.ts

import { Queue, MemoryStorage } from '@platformatic/job-queue'

// Create a queue with in-memory storage
const storage = new MemoryStorage()
const queue = new Queue<{ email: string }, { sent: boolean }>({
  storage,
  concurrency: 5
})

// Register a job handler
queue.execute(async (job) => {
  console.log(`Processing job ${job.id}:`, job.payload)
  // Send email...
  return { sent: true }
})

// Start the queue
await queue.start()

// Enqueue jobs
await queue.enqueue('email-1', { email: '[e-mail]' })

// Optional per-job TTL override for cached result/error
await queue.enqueue('email-ttl', { email: '[e-mail]' }, {
  resultTTL: 5 * 60 * 1000 // 5 minutes
})

// Or wait for the result
const result = await queue.enqueueAndWait('email-2', { email: '[e-mail]' }, {
  timeout: 30000,
  resultTTL: 24 * 60 * 60 * 1000 // keep this result for 24h
})
console.log('Result:', result) // { sent: true }

// Graceful shutdown
await queue.stop()

Run the Example

Save the code above as index.ts and run it:

node --experimental-strip-types index.ts

Expected Output

Queue started and ready to process jobs
Processing job email-1: { email: '[e-mail]' }
Job enqueued
Processing job email-ttl: { email: '[e-mail]' }
Processing job email-2: { email: '[e-mail]' }
Result: { sent: true }
Queue stopped gracefully

Understanding the Flow

How enqueue() works

When you call enqueue(id, payload, options?), the library:

1. Checks if a job with that ID already exists
2. If it’s new, serializes the payload and adds it to the queue
3. Returns immediately with status 'queued', 'duplicate', or 'completed'
4. The consumer picks up the job asynchronously

How enqueueAndWait() works

When you call enqueueAndWait(id, payload, options?), the library:

1. Enqueues the job (same as enqueue())
2. Subscribes to completion events for that job ID
3. Waits until the job completes or times out
4. Returns the result or throws an error

This is perfect for request/response patterns across services.

How deduplication works

Jobs are identified by their ID. When you enqueue a job:

• If no job with that ID exists, it’s queued
• If a job is queued/processing, you get status 'duplicate'
• If a job completed, you get the cached result immediately

Results are cached for resultTTL milliseconds (default: 1 hour).

How graceful shutdown works

When you call stop():

1. The queue stops accepting new jobs
2. In-flight jobs continue processing
3. The queue waits up to visibilityTimeout for jobs to complete
4. Any incomplete jobs are returned to the queue
5. Storage is disconnected

Configuration Options

The Queue constructor accepts these options:

Option	Type	Default	Description
storage	Storage	required	Storage backend instance
workerId	string	uuid()	Unique identifier for this worker
concurrency	number	1	Number of jobs to process in parallel
maxRetries	number	3	Maximum retry attempts for failed jobs
blockTimeout	number	5	Seconds to wait when polling for jobs
visibilityTimeout	number	30000	Milliseconds before a processing job is considered stalled
resultTTL	number	3600000	Milliseconds to cache job results (1 hour)
logger	pino.Logger	abstractLogger	Logger instance

Error Handling

Handle errors using try-catch with typed error classes:

import { TimeoutError, JobFailedError } from '@platformatic/job-queue'

try {
  const result = await queue.enqueueAndWait('job-1', payload, {
    timeout: 10000
  })
  console.log('Got result:', result)
} catch (error) {
  if (error instanceof TimeoutError) {
    console.log('Request timed out')
  } else if (error instanceof JobFailedError) {
    console.log('Job failed:', error.originalError)
  }
}

Jobs that fail will be retried up to maxRetries times with exponential backoff. After all retries are exhausted, they enter the 'failed' state.

Listening to Events

The queue emits events for monitoring:

// Job events
queue.on('enqueued', (id) => {
  console.log(`Job ${id} was enqueued`)
})

queue.on('completed', (id, result) => {
  console.log(`Job ${id} completed:`, result)
})

queue.on('failed', (id, error) => {
  console.log(`Job ${id} failed:`, error.message)
})

queue.on('failing', (id, error, attempt) => {
  console.log(`Job ${id} failed attempt ${attempt}, will retry:`, error.message)
})

// Lifecycle events
queue.on('started', () => {
  console.log('Queue started')
})

queue.on('stopped', () => {
  console.log('Queue stopped')
})

// Error events
queue.on('error', (error) => {
  console.error('Queue error:', error)
})

Next Steps

Now that you have a working queue, explore more advanced features:

Storage Backends

Learn about Redis, FileStorage, and MemoryStorage

Queue Configuration

Configure retries, timeouts, and concurrency

Stalled Job Recovery

Set up the Reaper for automatic recovery

Producer/Consumer

Separate producers and consumers across services