Jobs

Bueno's job queue lets you run background tasks reliably, with retries, scheduling, and prioritization.

Creating a job

import { defineJob } from '@buenojs/bueno/jobs';

const sendWelcomeEmail = defineJob('send-welcome-email', async (payload: { userId: number }) => {
  const user = await db.query('SELECT * FROM users WHERE id = $1', [payload.userId]);
  await mailer.send({
    to: user.email,
    subject: 'Welcome!',
    html: await templates.render('welcome', { name: user.name }),
  });
});

Registering and starting the queue

import { createQueue } from '@buenojs/bueno/jobs';

const queue = createQueue({
  concurrency: 5,
  redis: Bun.env.REDIS_URL,
});

queue.register(sendWelcomeEmail);
queue.register(processUpload);
queue.register(generateReport);

await queue.start();

Enqueuing jobs

// Enqueue immediately
await queue.enqueue('send-welcome-email', { userId: 42 });

// Enqueue with options
await queue.enqueue('generate-report', { reportId: 7 }, {
  priority: 10,         // Higher = processed first
  delay: 5000,          // Wait 5 seconds before processing
  maxRetries: 3,        // Retry up to 3 times on failure
  retryDelay: 10_000,   // Wait 10 seconds between retries
});

Scheduled jobs (cron)

import { scheduleCron } from '@buenojs/bueno/jobs';

// Run every day at 2am
scheduleCron('daily-cleanup', '0 2 * * *', async () => {
  await db.execute('DELETE FROM sessions WHERE expires_at < NOW()');
});

// Run every 5 minutes
scheduleCron('sync-inventory', '*/5 * * * *', async () => {
  await syncInventoryFromWarehouse();
});

Job events

queue.on('job:started', ({ name, payload }) => {
  console.log(`Starting ${name}`, payload);
});

queue.on('job:completed', ({ name, duration }) => {
  console.log(`Completed ${name} in ${duration}ms`);
});

queue.on('job:failed', ({ name, error, attempt }) => {
  console.error(`Failed ${name} (attempt ${attempt})`, error);
});

Queue configuration

interface QueueConfig {
  concurrency?: number;    // Max concurrent jobs (default: 10)
  redis?: string;          // Redis URL for persistence
  prefix?: string;         // Queue key prefix
  pollInterval?: number;   // MS between polling when idle
}

Job retries

const fragileJob = defineJob('fragile', async (payload) => {
  const res = await fetch(externalApi);
  if (!res.ok) throw new Error(`API returned ${res.status}`);
  // ...
});

await queue.enqueue('fragile', payload, {
  maxRetries: 5,
  retryDelay: 30_000,  // 30 second backoff
});

Graceful shutdown

process.on('SIGTERM', async () => {
  await queue.stop();  // Wait for running jobs to finish
  process.exit(0);
});