Background Jobs

Patterns for background processing, task queues, and scheduled work.

Email Queue Pattern

Queue tasks from a route handler and process them in a background job:

// api/invites.js
export const POST = {
  handler: async (ctx) => {
    const { queue } = ctx;
    const { emails, message } = ctx.req.body;

    for (const email of emails) {
      await queue.push('send-invite', { email, message, invitedBy: ctx.req.session.userId });
    }

    return { data: { queued: emails.length } };
  },
};

// jobs/send-invite.js
export default {
  name: 'send-invite',
  schedule: 'continuous',
  retries: 3,
  throughput: { maxPerMinute: 60 },
  handler: async (ctx) => {
    const { queue, db, mail, log } = ctx;
    const items = await queue.pop('send-invite', 5);
    if (items.length === 0) {
      await new Promise((r) => setTimeout(r, 5000));
      return;
    }

    for (const item of items) {
      try {
        const inviter = await db('users').where('id', item.data.invitedBy).first();
        await mail.send({
          to: item.data.email,
          subject: `${inviter.name} invited you!`,
          html: `<p>${item.data.message}</p>`,
        });
        await queue.remove([item.id]);
        log.info('Invite sent', { email: item.data.email });
      } catch (err) {
        log.error('Failed to send invite', { email: item.data.email, error: err.message });
        // Leave in queue — lock expires and it will be retried
      }
    }
  },
};

Batch Queue Inserts

Use pushBulk to queue many items in one database round-trip:

// api/notifications/broadcast.js
export const POST = {
  schema: {
    body: {
      userIds: 'string[]',
      message: 'string>0',
    },
  },
  handler: async (ctx) => {
    const { queue } = ctx;
    const { userIds, message } = ctx.req.body;

    // All items queued in a single batch insert
    await queue.pushBulk(
      'send-notification',
      userIds.map((userId) => ({ userId, message })),
    );

    return { data: { queued: userIds.length } };
  },
};

Scheduled Reports

Run a job on a cron schedule:

// jobs/daily-report.js
export default {
  name: 'daily-report',
  schedule: '0 8 * * *', // Every day at 8am
  handler: async (ctx) => {
    const { db, mail } = ctx;
    const yesterday = new Date(Date.now() - 86400000).toISOString().split('T')[0];

    const [signups, orders] = await Promise.all([
      db('users').whereRaw('DATE(created_at) = ?', [yesterday]).count('* as count').first(),
      db('orders')
        .whereRaw('DATE(created_at) = ?', [yesterday])
        .sum('total as revenue')
        .count('* as count')
        .first(),
    ]);

    await mail.send({
      to: 'team@myapp.com',
      subject: `Daily Report — ${yesterday}`,
      html: `
        <h2>Daily Report for ${yesterday}</h2>
        <p>New signups: ${signups.count}</p>
        <p>Orders: ${orders.count} (Revenue: $${orders.revenue || 0})</p>
      `,
    });
  },
};

Database Cleanup Job

Periodically purge old or expired records:

// jobs/cleanup.js
export default {
  name: 'cleanup',
  schedule: '0 2 * * *', // Every night at 2am
  handler: async (ctx) => {
    const { db, log } = ctx;
    const cutoff = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000).toISOString(); // 30 days ago

    const [expiredSessions, oldNotifications] = await Promise.all([
      db('sessions').where('expires_at', '<', cutoff).delete(),
      db('notifications').where('created_at', '<', cutoff).where('read', true).delete(),
    ]);

    log.info('Cleanup complete', { expiredSessions, oldNotifications });
  },
};

Continuous Worker with Cooldown

A continuous job that processes items and sleeps when the queue is empty. Use cooldownMs to avoid hammering the database between runs:

// jobs/process-exports.js
export default {
  name: 'process-exports',
  schedule: 'continuous',
  cooldownMs: 2000, // Wait 2s between runs
  handler: async (ctx) => {
    const { queue, db, files, log } = ctx;
    const items = await queue.pop('export-request', 1);
    if (items.length === 0) return;

    const [item] = items;
    try {
      const rows = await db(item.data.table).select('*');
      const csv = [Object.keys(rows[0]).join(','), ...rows.map((r) => Object.values(r).join(','))].join('\n');

      await files.write(`exports/${item.data.exportId}.csv`, Buffer.from(csv));
      await db('exports').where('id', item.data.exportId).update({ status: 'done' });
      await queue.remove([item.id]);
      log.info('Export complete', { exportId: item.data.exportId });
    } catch (err) {
      log.error('Export failed', { exportId: item.data.exportId, error: err.message });
      await db('exports').where('id', item.data.exportId).update({ status: 'failed' });
      await queue.remove([item.id]);
    }
  },
};