Worker

// BullMQ를 사용한 Node.js Worker 예제

// queue.js - 큐 설정
import { Queue } from 'bullmq';

const connection = {
  host: process.env.REDIS_HOST || 'localhost',
  port: 6379,
};

// 이메일 발송 큐
export const emailQueue = new Queue('email', { connection });

// 이미지 처리 큐
export const imageQueue = new Queue('image-processing', {
  connection,
  defaultJobOptions: {
    attempts: 3,                    // 최대 3번 재시도
    backoff: {
      type: 'exponential',          // 지수 백오프
      delay: 1000,                  // 1초부터 시작
    },
    removeOnComplete: 100,          // 완료된 작업 100개만 유지
    removeOnFail: 500,              // 실패한 작업 500개만 유지
  },
});

// producer.js - 작업 추가
export async function sendWelcomeEmail(userId, email) {
  const job = await emailQueue.add('welcome', {
    userId,
    email,
    template: 'welcome',
  }, {
    priority: 1,                    // 높은 우선순위
    delay: 0,                       // 즉시 실행
    jobId: `welcome-${userId}`,     // 중복 방지용 ID
  });

  console.log(`Job ${job.id} added to queue`);
  return job;
}

export async function processImage(imageId, operations) {
  return imageQueue.add('resize', {
    imageId,
    operations,  // [{ type: 'resize', width: 800 }, { type: 'compress' }]
  });
}

// worker.js - Worker 정의
import { Worker } from 'bullmq';
import { sendEmail } from './services/email';
import { resizeImage, compressImage } from './services/image';

// 이메일 Worker
const emailWorker = new Worker('email', async (job) => {
  console.log(`Processing email job ${job.id}`);

  const { userId, email, template } = job.data;

  // 멱등성: 이미 발송된 이메일인지 확인
  if (await isEmailAlreadySent(userId, template)) {
    console.log(`Email already sent to ${email}`);
    return { status: 'skipped', reason: 'already_sent' };
  }

  await sendEmail({
    to: email,
    template,
    data: { userId },
  });

  await markEmailAsSent(userId, template);

  return { status: 'sent', email };
}, {
  connection,
  concurrency: 5,  // 동시 처리 수
});

// 이벤트 핸들링
emailWorker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);
});

emailWorker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed:`, err.message);
  // 알림 발송, 메트릭 기록 등
});

// 이미지 처리 Worker
const imageWorker = new Worker('image-processing', async (job) => {
  const { imageId, operations } = job.data;

  let image = await loadImage(imageId);

  for (const op of operations) {
    // 진행률 업데이트
    await job.updateProgress(operations.indexOf(op) / operations.length * 100);

    if (op.type === 'resize') {
      image = await resizeImage(image, op.width, op.height);
    } else if (op.type === 'compress') {
      image = await compressImage(image, op.quality);
    }
  }

  const outputPath = await saveImage(image);
  return { outputPath };
}, {
  connection,
  concurrency: 2,  // 리소스 많이 쓰므로 제한
});

// Kubernetes Deployment (worker.yaml)
/*
apiVersion: apps/v1
kind: Deployment
metadata:
  name: email-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: email-worker
  template:
    spec:
      containers:
      - name: worker
        image: myapp/worker:v1.0.0
        command: ["node", "worker.js"]
        env:
        - name: REDIS_HOST
          valueFrom:
            configMapKeyRef:
              name: redis-config
              key: host
        resources:
          requests:
            cpu: 100m
            memory: 256Mi
*/