Nuda Kit

Backend Development

Queues & Jobs

Background job processing with BullMQ.

Nuda Kit uses BullMQ directly for background job processing. Jobs are processed asynchronously using Redis as the message broker. A thin BaseJob class (app/jobs/base_job.ts) provides the job structure, dispatching, and a registry the worker uses to route jobs to their handlers.

Configuration

Queue configuration is in config/queue.ts:

import env from '#start/env'
import type { ConnectionOptions, DefaultJobOptions } from 'bullmq'

const queueConfig: QueueConfig = {
  connection: {
    host: env.get('QUEUE_REDIS_HOST'),
    port: env.get('QUEUE_REDIS_PORT'),
    password: env.get('QUEUE_REDIS_PASSWORD'),
  },

  // Queues the worker processes by default
  queues: ['emails', 'user_deletions'],

  defaultJobOptions: {
    attempts: 3,                                    // Total attempts before a job is marked failed
    backoff: { type: 'exponential', delay: 1000 }, // Wait between retries
    removeOnComplete: 100,                          // Keep last 100 completed jobs
    removeOnFail: 100,                              // Keep last 100 failed jobs
  },

  worker: {
    concurrency: 5, // Jobs processed in parallel per queue
    stalledInterval: 30_000,
    maxStalledCount: 1,
    lockDuration: 30_000,
  },
}

export default queueConfig

Environment variables:

QUEUE_REDIS_HOST=localhost
QUEUE_REDIS_PORT=6379
QUEUE_REDIS_PASSWORD=

Running Queue Workers

To process jobs, start the worker with the queue:work ace command. A single worker process can handle all queues:

# Process all configured queues (from config/queue.ts)
node ace queue:work

Process Specific Queues

Pass queue names as arguments to dedicate a process to specific queues:

# Only the emails queue
node ace queue:work emails

# Multiple queues in one process
node ace queue:work emails user_deletions

Concurrency

Override how many jobs are processed in parallel per queue with -c:

node ace queue:work emails -c 10

The worker logs every completed and failed job:

[ info ] [emails] SendMagicLinkEmailJob#556 completed
[ error ] [emails] SendVerificationEmailJob#557 failed (attempt 1/3): Row not found
Queue workers must be running for jobs to be processed. In production, use a process manager like PM2 or supervisor to keep workers alive.

Existing Queues

Nuda Kit comes with two pre-configured queues:

QueuePurposeJobs
emailsEmail deliveryVerification, password reset, magic links, invitations
user_deletionsAccount cleanupDelete user data, teams, and files

Existing Jobs

Jobs are located in app/jobs/:

JobQueueDescription
SendVerificationEmailJobemailsSends email verification link
SendResetPasswordEmailJobemailsSends password reset email
SendMagicLinkEmailJobemailsSends magic link for passwordless login
SendInvitationEmailJobemailsSends team invitation email
DeleteUserJobuser_deletionsDeletes user, owned teams, and uploaded files

Job Structure

Jobs extend BaseJob, declare their queue in static options, and implement an execute method. The payload is available as this.payload:

import BaseJob from '#jobs/base_job'
import type { JobOptions } from '#jobs/base_job'

export interface SendVerificationEmailPayload {
  userId: number
}

export default class SendVerificationEmailJob extends BaseJob<SendVerificationEmailPayload> {
  static options: JobOptions = {
    name: 'SendVerificationEmailJob', // Must match the class name
    queue: 'emails',                  // Queue the job is dispatched to
  }

  // Main job logic
  async execute(): Promise<void> {
    const user = await User.findOrFail(this.payload.userId)
    await mail.send(new VerifyEmailNotification(user))
  }

  // Optional: called when all retries are exhausted
  async failed(error: Error): Promise<void> {
    // Log failure, notify admin, etc.
  }
}

Job Options

OptionDefaultDescription
name—Unique job name, used by the worker to find the handler
queue—Queue the job is dispatched to
maxRetriesfrom config (attempts: 3 total)Retries after the first attempt
timeoutnoneMax execution time, e.g. '30s', '5m'
removeOnCompletefrom configCompleted jobs to keep
removeOnFailfrom configFailed jobs to keep, or { age: '7d' }

Dispatching Jobs

Call the static dispatch() method on the job class. The queue is taken from the job's options, so call sites stay clean:

import SendVerificationEmailJob from '#jobs/send_verification_email_job'

await SendVerificationEmailJob.dispatch({ userId: user.id })

Delayed Jobs

Chain .in() to delay processing:

// Process after 10 minutes
await SendReminderJob.dispatch({ userId: user.id }).in('10m')

// Or in milliseconds
await SendReminderJob.dispatch({ userId: user.id }).in(5000)

Creating New Jobs

1. Create the Job File

Create a new file in app/jobs/:

// app/jobs/send_welcome_email_job.ts

import BaseJob from '#jobs/base_job'
import type { JobOptions } from '#jobs/base_job'
import mail from '@adonisjs/mail/services/main'
import User from '#models/user'
import WelcomeEmail from '#mails/welcome_email'

export interface SendWelcomeEmailPayload {
  userId: number
}

export default class SendWelcomeEmailJob extends BaseJob<SendWelcomeEmailPayload> {
  static options: JobOptions = {
    name: 'SendWelcomeEmailJob',
    queue: 'emails',
  }

  async execute(): Promise<void> {
    const user = await User.findOrFail(this.payload.userId)
    await mail.send(new WelcomeEmail(user))
  }

  async failed(error: Error): Promise<void> {
    console.error(`Failed to send welcome email to user ${this.payload.userId}:`, error)
  }
}

2. Register the Job

Add the job to the registry in app/jobs/index.ts so the worker can find it:

import SendWelcomeEmailJob from '#jobs/send_welcome_email_job'

const jobs = [
  // ... existing jobs
  SendWelcomeEmailJob,
] as const
Unregistered jobs fail in the worker with No registered handler for job "...". Don't skip this step.

3. Dispatch the Job

import SendWelcomeEmailJob from '#jobs/send_welcome_email_job'

await SendWelcomeEmailJob.dispatch({ userId: user.id })

Creating New Queues

Add the queue name to config/queue.ts and reference it from a job's options:

// config/queue.ts
queues: ['emails', 'user_deletions', 'notifications'],

// app/jobs/send_push_notification_job.ts
static options: JobOptions = {
  name: 'SendPushNotificationJob',
  queue: 'notifications',
}

The default node ace queue:work picks it up automatically, or start a dedicated worker:

node ace queue:work notifications

Queue Organization Tips

QueueUse Case
emailsEmail delivery (SMTP can be slow)
notificationsPush notifications, SMS
exportsLarge file exports, reports
importsData imports, CSV processing
cleanupScheduled maintenance tasks

Event-Driven Jobs

Jobs are often dispatched from event listeners. This keeps controllers clean:

// app/listeners/send_verification_email.ts

import { inject } from '@adonisjs/core'
import UserCreated from '#events/user_created'
import SendVerificationEmailJob from '#jobs/send_verification_email_job'

@inject()
export default class SendVerificationEmail {
  async handle(event: UserCreated) {
    await SendVerificationEmailJob.dispatch({ userId: event.user.id })
  }
}

Job Lifecycle

┌─────────────────────────────────────────────────────────┐
│                     Job Dispatched                       │
└─────────────────────────┬───────────────────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────┐
│                    Added to Queue                        │
│                 (stored in Redis)                        │
└─────────────────────────┬───────────────────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────┐
│                  Worker Picks Up Job                     │
│             (queue:work must be running)                 │
└─────────────────────────┬───────────────────────────────┘
                          ▼
                    ┌─────────────┐
                    │  execute()  │
                    │   called    │
                    └──────┬──────┘
                           │
              ┌────────────┴────────────┐
              ▼                         ▼
        ┌──────────┐              ┌──────────┐
        │ Success  │              │  Failed  │
        └────┬─────┘              └────┬─────┘
             │                         │
             ▼                    ┌────┴────┐
        ┌─────────┐               ▼         ▼
        │Completed│          ┌────────┐ ┌────────┐
        └─────────┘          │ Retry  │ │failed()│
                             │(≤3x)   │ │called  │
                             └────────┘ └────────┘
The queue:work command commits the router before starting, so jobs can build route URLs (e.g. signed magic link URLs in emails) with signedUrlFor().

Testing Jobs

Use the queue fake helper to test job dispatching without actually processing:

import { test } from '@japa/runner'
import QueueFake from '#tests/helpers/queue_fake'
import SendVerificationEmailJob from '#jobs/send_verification_email_job'

test('dispatches verification email job', async ({ assert, cleanup }) => {
  await QueueFake.fake()
  cleanup(async () => await QueueFake.restore())

  // ... create user ...

  const jobs = QueueFake.getDispatchedJobs()

  assert.lengthOf(jobs, 1)
  assert.equal(jobs[0].jobClass, SendVerificationEmailJob)
  assert.equal(jobs[0].options?.queueName, 'emails')
})

Production Setup

PM2 Configuration

// ecosystem.config.js
module.exports = {
  apps: [
    {
      name: 'api',
      script: 'build/bin/server.js',
    },
    {
      name: 'queue-emails',
      script: 'node',
      args: 'ace queue:work emails',
      cwd: './build',
    },
    {
      name: 'queue-deletions',
      script: 'node',
      args: 'ace queue:work user_deletions',
      cwd: './build',
    },
  ],
}

Scaling Workers

You can run multiple workers per queue for higher throughput:

# Run 3 email workers
pm2 start queue-emails -i 3

Or increase concurrency within a single process:

node ace queue:work emails -c 20

Resources

Database

PostgreSQL database with Lucid ORM.

Emails

Transactional email with MJML templates.