작업 큐: BullMQ + Redis

pnpm add bullmq ioredis

// queues/email.queue.ts
import { Queue } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis(process.env.REDIS_URL!, {
  maxRetriesPerRequest: null,
});

export const emailQueue = new Queue('email', { connection });

export async function sendWelcomeEmail(userId: string, email: string) {
  await emailQueue.add(
    'welcome',
    { userId, email },
    {
      attempts: 3,
      backoff: { type: 'exponential', delay: 5000 },
      removeOnComplete: 100,    // 완료된 잡 100개만 보관
      removeOnFail: 500,
    }
  );
}

// workers/email.worker.ts
import { Worker } from 'bullmq';
import { sendEmail } from '@/lib/email';

const worker = new Worker(
  'email',
  async (job) => {
    const { type, userId, email } = job.data;
    await job.log(`Processing ${type} for ${email}`);

    if (type === 'welcome') {
      await sendEmail(email, '환영합니다', `<h1>가입을 환영합니다</h1>`);
    } else if (type === 'reset-password') {
      await sendResetEmail(job.data);
    }

    return { sent: true, at: new Date() };
  },
  {
    connection,
    concurrency: 5,  // 동시 5개 처리
    limiter: {
      max: 100,        // 분당 최대 100개 (외부 API 한도)
      duration: 60000,
    },
  }
);

worker.on('completed', (job) => console.log(`Job ${job.id} done`));
worker.on('failed', (job, err) => console.error(`Job ${job?.id} failed:`, err));

// 즉시 (긴급)
await queue.add('urgent', data, { priority: 1 });

// 5분 후
await queue.add('reminder', data, { delay: 5 * 60 * 1000 });

// 매일 오전 9시 (cron)
import { JobScheduler } from 'bullmq';
const scheduler = new JobScheduler('email');
await scheduler.upsertJobScheduler(
  'daily-digest',
  { pattern: '0 9 * * *' },
  { name: 'digest', data: {} }
);


// 1시간 후 (지연 알림)
await queue.add(
  'reminder',
  { userId, message: '결제 완료해주세요' },
  { delay: 60 * 60 * 1000 }
);

import sharp from 'sharp';
import { GetObjectCommand, PutObjectCommand } from '@aws-sdk/client-s3';

const imageWorker = new Worker(
  'image',
  async (job) => {
    const { srcKey, sizes } = job.data;

    const original = await s3.send(new GetObjectCommand({
      Bucket: BUCKET, Key: srcKey,
    }));
    const buffer = Buffer.from(await original.Body!.transformToByteArray());

    const results = [];
    for (const size of sizes) {
      const resized = await sharp(buffer)
        .resize(size.width, size.height, { fit: 'cover' })
        .webp({ quality: 80 })
        .toBuffer();

      const dstKey = srcKey.replace(/\.\w+$/, `_${size.width}x${size.height}.webp`);
      await s3.send(new PutObjectCommand({
        Bucket: BUCKET, Key: dstKey, Body: resized, ContentType: 'image/webp',
      }));
      results.push(dstKey);

      await job.updateProgress((results.length / sizes.length) * 100);
    }

    return { resized: results };
  },
  { connection, concurrency: 3 }
);


// 사용
await imageQueue.add('resize', {
  srcKey: 'uploads/abc.jpg',
  sizes: [
    { width: 200, height: 200 },
    { width: 800, height: 600 },
  ],
});

const webhookWorker = new Worker(
  'webhook',
  async (job) => {
    const { url, payload, secret } = job.data;
    const signature = crypto.createHmac('sha256', secret)
      .update(JSON.stringify(payload))
      .digest('hex');

    const res = await fetch(url, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-Signature': signature,
      },
      body: JSON.stringify(payload),
      signal: AbortSignal.timeout(10000),
    });

    if (!res.ok) {
      throw new Error(`Webhook failed: ${res.status}`);
    }
    return { status: res.status };
  },
  {
    connection,
    concurrency: 10,
  }
);


// 발송 — 자동 재시도
await webhookQueue.add('order.created', payload, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 2000 },  // 2s, 4s, 8s, 16s, 32s
});

import { ExpressAdapter } from '@bull-board/express';
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(imageQueue),
    new BullMQAdapter(webhookQueue),
  ],
  serverAdapter,
});

app.use('/admin/queues', requireAdmin, serverAdapter.getRouter());