Queues

node ace add @adonisjs/queue

import env from '#start/env'
import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: env.get('QUEUE_DRIVER', 'redis'),

  adapters: {
    redis: drivers.redis({
      connectionName: 'main',
    }),
    sync: drivers.sync(),
  },

  worker: {
    concurrency: 5,
    idleDelay: '2s',
  },

  locations: ['./app/jobs/**/*.ts'],
})

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'redis',
  adapters: {
    redis: drivers.redis({
      // Uses the 'main' connection from config/redis.ts
      connectionName: 'main',
    }),
  },
  // ...
})

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'database',
  adapters: {
    database: drivers.database({
      connectionName: 'primary',
    }),
  },
  // ...
})

import { BaseSchema } from '@adonisjs/lucid/schema'
import { QueueSchemaService } from '@boringnode/queue'

export default class extends BaseSchema {
  async up() {
    const schemaService = new QueueSchemaService(this.db.connection().getWriteClient())

    await schemaService.createJobsTable()
    await schemaService.createSchedulesTable()
  }

  async down() {
    const schemaService = new QueueSchemaService(this.db.connection().getWriteClient())

    await schemaService.dropSchedulesTable()
    await schemaService.dropJobsTable()
  }
}

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'sync',
  adapters: {
    sync: drivers.sync(),
  },
  // ...
})

node ace make:job process_payment

import { Job } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'

interface ProcessPaymentPayload {
  orderId: number
  amount: number
  currency: string
}

export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  static options: JobOptions = {
    queue: 'default',
    maxRetries: 3,
  }

  async execute() {
    const { orderId, amount, currency } = this.payload

    // Process the payment using your payment gateway
    console.log(`Processing payment of ${amount} ${currency} for order ${orderId}`)
  }

  async failed(error: Error) {
    console.error(`Payment processing failed for order ${this.payload.orderId}:`, error.message)
  }
}

export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  async execute() {
    // Access job metadata
    console.log(`Job ID: ${this.context.jobId}`)
    console.log(`Attempt: ${this.context.attempt}`)
    console.log(`Queue: ${this.context.queue}`)
    console.log(`Priority: ${this.context.priority}`)
  }
}

import { Job } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'

interface GenerateReportPayload {
  reportId: number
  rows: number[]
}

export default class GenerateReport extends Job<GenerateReportPayload> {
  static options: JobOptions = {
    timeout: '5m',
    failOnTimeout: false, // Retry instead of failing permanently
  }

  async execute() {
    for (const rowId of this.payload.rows) {
      // Check if the job has timed out before processing each row
      if (this.signal?.aborted) {
        throw new Error('Job timed out during report generation')
      }

      await this.processRow(rowId)
    }
  }

  private async processRow(rowId: number) {
    // Process individual row
  }
}

import { inject } from '@adonisjs/core'
import { Job } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'
import PaymentService from '#services/payment_service'

interface ProcessPaymentPayload {
  orderId: number
  amount: number
  currency: string
}

@inject()
export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  static options: JobOptions = {
    queue: 'payments',
    maxRetries: 3,
  }

  constructor(private paymentService: PaymentService) {
    super()
  }

  async execute() {
    await this.paymentService.charge(
      this.payload.orderId,
      this.payload.amount,
      this.payload.currency
    )
  }
}

import ProcessPayment from '#jobs/process_payment'

export default class OrdersController {
  async store({ request, response }: HttpContext) {
    const order = await Order.create(request.body())

    // Dispatch the job for background processing
    await ProcessPayment.dispatch({
      orderId: order.id,
      amount: order.total,
      currency: 'USD',
    })

    return response.created(order)
  }
}

// Dispatch to a specific queue with high priority
await ProcessPayment.dispatch({
  orderId: order.id,
  amount: order.total,
  currency: 'USD',
})
  .toQueue('payments')
  .priority(1)

import SendNewsletter from '#jobs/send_newsletter'

export default class NewsletterService {
  async sendToAllSubscribers(subscribers: { email: string }[]) {
    const payloads = subscribers.map((subscriber) => ({
      email: subscriber.email,
      subject: 'Monthly Newsletter',
    }))

    const { jobIds } = await SendNewsletter.dispatchMany(payloads)
      .toQueue('emails')
      .group('newsletter-march-2025')

    console.log(`Dispatched ${jobIds.length} newsletter jobs`)
  }
}

import { defineConfig, drivers } from '@adonisjs/queue'
import { exponentialBackoff } from '@adonisjs/queue'

export default defineConfig({
  default: 'redis',
  adapters: {
    redis: drivers.redis({ connectionName: 'main' }),
  },
  retry: {
    maxRetries: 3,
    backoff: exponentialBackoff(),
  },
  // ...
})

import { exponentialBackoff } from '@adonisjs/queue'

// Default: 1s base, 5m max, 2x multiplier, jitter enabled
exponentialBackoff()

// Custom
exponentialBackoff({ baseDelay: '500ms', maxDelay: '1m' })

import { linearBackoff } from '@adonisjs/queue'

// Default: 5s base, 2m max
linearBackoff()

// Custom
linearBackoff({ baseDelay: '10s', maxDelay: '5m' })

import { fixedBackoff } from '@adonisjs/queue'

// Default: 10s
fixedBackoff()

// Custom
fixedBackoff('30s')

import { customBackoff } from '@adonisjs/queue'

customBackoff({
  strategy: 'exponential',
  baseDelay: '100ms',
  maxDelay: '30s',
  multiplier: 3,
  jitter: false,
})

import { Job } from '@adonisjs/queue'
import { exponentialBackoff } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'

export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  static options: JobOptions = {
    maxRetries: 5,
    retry: {
      backoff: exponentialBackoff({ baseDelay: '2s', maxDelay: '10m' }),
    },
  }

  async execute() {
    // Payment processing logic
  }
}

import CleanupExpiredSessions from '#jobs/cleanup_expired_sessions'
import GenerateWeeklyReport from '#jobs/generate_weekly_report'

// Run every night at midnight
await CleanupExpiredSessions.schedule({ retentionDays: 30 })
  .cron('0 0 * * *')
  .timezone('Europe/Paris')

// Run every Monday at 9 AM
await GenerateWeeklyReport.schedule({ type: 'weekly' })
  .cron('0 9 * * MON')
  .timezone('America/New_York')

import SyncInventory from '#jobs/sync_inventory'

// Run every 5 minutes
await SyncInventory.schedule({ source: 'warehouse-api' })
  .every('5m')

import { Schedule } from '@adonisjs/queue'

export default class AdminController {
  async listSchedules() {
    // List all schedules
    const schedules = await Schedule.list()

    // List only active schedules
    const active = await Schedule.list({ status: 'active' })

    return schedules
  }

  async pauseSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      await schedule.pause()
    }
  }

  async resumeSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      await schedule.resume()
    }
  }

  async triggerSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      // Immediately dispatch the scheduled job
      await schedule.trigger()
    }
  }

  async deleteSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      await schedule.delete()
    }
  }
}

# List all scheduled jobs
node ace queue:scheduler:list

# List only active schedules
node ace queue:scheduler:list --status=active

# Remove a specific schedule
node ace queue:scheduler:remove <schedule-id>

# Remove all schedules
node ace queue:scheduler:clear

node ace queue:work

# Process specific queues
node ace queue:work --queue=payments,emails

# Set concurrency (number of jobs processed simultaneously)
node ace queue:work --concurrency=10

export default defineConfig({
  // ...
  worker: {
    concurrency: 5,
    idleDelay: '2s',
    stalledThreshold: '30s',
    stalledInterval: '30s',
    maxStalledCount: 1,
    gracefulShutdown: true,
  },
})

import { test } from '@japa/runner'
import { QueueManager } from '@adonisjs/queue'
import ProcessPayment from '#jobs/process_payment'

test.group('Orders', (group) => {
  group.each.teardown(() => {
    QueueManager.restore()
  })

  test('dispatches a payment job when creating an order', async ({ client }) => {
    const fake = QueueManager.fake()

    const response = await client.post('/orders').json({
      product_id: 1,
      quantity: 2,
    })

    response.assertStatus(201)

    // Assert the job was dispatched
    fake.assertPushed(ProcessPayment)

    // Assert with payload matching
    fake.assertPushed(ProcessPayment, {
      payload: { orderId: 1, amount: 100, currency: 'USD' },
    })
  })

  test('does not dispatch a payment job for free orders', async ({ client }) => {
    const fake = QueueManager.fake()

    await client.post('/orders').json({
      product_id: 1,
      quantity: 0,
    })

    fake.assertNotPushed(ProcessPayment)
  })
})

import { test } from '@japa/runner'
import { QueueManager } from '@adonisjs/queue'
import SendNewsletter from '#jobs/send_newsletter'

test('dispatches newsletter jobs for all subscribers', async ({ client }) => {
  const fake = QueueManager.fake()

  await client.post('/newsletters/send')

  // Match with a function
  fake.assertPushed(SendNewsletter, {
    payload: (payload) => payload.email.endsWith('@example.com'),
  })

  // Check delay
  fake.assertPushed(SendNewsletter, {
    delay: (delay) => delay !== undefined && delay > 0,
  })
})