# Background Jobs & Queues mit BullMQ

**Meta-Description:** Background Job Processing mit BullMQ und Redis. Queues, Workers, Retries und Job Scheduling für Node.js.

**Keywords:** BullMQ, Background Jobs, Queue Processing, Redis Queue, Job Scheduling, Worker Threads, Async Processing

---

## Einführung

Nicht jede Aufgabe sollte im Request-Response-Zyklus laufen. **Background Jobs** ermöglichen Email-Versand, Image Processing, Reports – alles asynchron mit Retries, Prioritäten und Scheduling.

---

## Warum Background Jobs?

```
┌─────────────────────────────────────────────────────────────┐
│              SYNCHRON vs ASYNCHRON                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Synchron (schlecht für):                                   │
│  ├── Email-Versand (3-5s)                                  │
│  ├── Image/Video Processing (10s+)                         │
│  ├── PDF Generation (2-5s)                                 │
│  ├── Externe API Calls (variabel)                          │
│  └── Batch Operations (Minuten)                            │
│                                                             │
│  User wartet... ⏳                                          │
│                                                             │
│  Asynchron (Background):                                    │
│  ├── Request → Response (50ms)                             │
│  ├── Job in Queue → Worker verarbeitet                     │
│  ├── Retries bei Fehlern                                   │
│  ├── Rate Limiting                                         │
│  └── Scheduling (Cron)                                     │
│                                                             │
│  User bekommt sofort Antwort! ✓                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘
```

---

## BullMQ Setup

```bash
npm install bullmq ioredis
```

```typescript
// lib/queue.ts
import { Queue, Worker, QueueEvents } from 'bullmq';
import IORedis from 'ioredis';

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

// Queue definieren
export const emailQueue = new Queue('email', { connection });
export const imageQueue = new Queue('image-processing', { connection });
export const reportQueue = new Queue('reports', { connection });

// Queue Events (für Monitoring)
export const emailQueueEvents = new QueueEvents('email', { connection });
```

---

## Job Types definieren

```typescript
// types/jobs.ts
export interface EmailJob {
  type: 'welcome' | 'reset-password' | 'notification';
  to: string;
  subject: string;
  template: string;
  data: Record<string, unknown>;
}

export interface ImageProcessingJob {
  imageId: string;
  operations: Array<{
    type: 'resize' | 'crop' | 'watermark';
    params: Record<string, unknown>;
  }>;
  outputFormat: 'webp' | 'avif' | 'jpeg';
}

export interface ReportJob {
  reportType: 'daily' | 'weekly' | 'monthly';
  userId: string;
  dateRange: {
    start: Date;
    end: Date;
  };
}
```

---

## Jobs hinzufügen

```typescript
// services/email.ts
import { emailQueue } from '@/lib/queue';
import { EmailJob } from '@/types/jobs';

export async function sendWelcomeEmail(userId: string, email: string) {
  const job = await emailQueue.add('send-email', {
    type: 'welcome',
    to: email,
    subject: 'Welcome to Our Platform!',
    template: 'welcome',
    data: { userId }
  } satisfies EmailJob, {
    // Job Options
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000  // 1s, 2s, 4s
    },
    removeOnComplete: {
      count: 1000  // Behalte letzte 1000 completed Jobs
    },
    removeOnFail: {
      count: 5000  // Behalte letzte 5000 failed Jobs
    }
  });

  return job.id;
}

// Mit Priorität
export async function sendUrgentEmail(data: EmailJob) {
  await emailQueue.add('send-email', data, {
    priority: 1  // Niedrigere Zahl = höhere Priorität
  });
}

// Verzögerter Job
export async function scheduleEmail(data: EmailJob, delay: number) {
  await emailQueue.add('send-email', data, {
    delay  // Millisekunden
  });
}

// Geplanter Job (Cron)
export async function scheduleReports() {
  await reportQueue.add('generate-report', {
    reportType: 'daily',
    // ...
  }, {
    repeat: {
      pattern: '0 8 * * *'  // Täglich um 8:00
    }
  });
}
```

---

## Worker implementieren

```typescript
// workers/email.worker.ts
import { Worker, Job } from 'bullmq';
import IORedis from 'ioredis';
import { Resend } from 'resend';
import { EmailJob } from '@/types/jobs';

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

const resend = new Resend(process.env.RESEND_API_KEY);

const emailWorker = new Worker<EmailJob>(
  'email',
  async (job: Job<EmailJob>) => {
    console.log(`Processing job ${job.id}: ${job.data.type}`);

    const { to, subject, template, data } = job.data;

    // Template rendern
    const html = await renderTemplate(template, data);

    // Email senden
    const result = await resend.emails.send({
      from: 'noreply@example.com',
      to,
      subject,
      html
    });

    console.log(`Email sent: ${result.id}`);

    return { emailId: result.id };
  },
  {
    connection,
    concurrency: 5,  // 5 Jobs parallel
    limiter: {
      max: 100,       // Max 100 Jobs
      duration: 60000 // Pro Minute (Rate Limiting)
    }
  }
);

// Event Handlers
emailWorker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed with result:`, result);
});

emailWorker.on('failed', (job, error) => {
  console.error(`Job ${job?.id} failed:`, error.message);
});

emailWorker.on('progress', (job, progress) => {
  console.log(`Job ${job.id} progress: ${progress}%`);
});

export { emailWorker };
```

```typescript
// workers/image.worker.ts
import { Worker, Job } from 'bullmq';
import sharp from 'sharp';
import { ImageProcessingJob } from '@/types/jobs';

const imageWorker = new Worker<ImageProcessingJob>(
  'image-processing',
  async (job: Job<ImageProcessingJob>) => {
    const { imageId, operations, outputFormat } = job.data;

    // Bild laden
    const image = await loadImage(imageId);
    let pipeline = sharp(image);

    // Operationen anwenden
    for (let i = 0; i < operations.length; i++) {
      const op = operations[i];

      // Progress updaten
      await job.updateProgress(Math.round((i / operations.length) * 100));

      switch (op.type) {
        case 'resize':
          pipeline = pipeline.resize(op.params.width, op.params.height);
          break;
        case 'crop':
          pipeline = pipeline.extract(op.params);
          break;
        case 'watermark':
          pipeline = pipeline.composite([{
            input: await loadWatermark(),
            gravity: 'southeast'
          }]);
          break;
      }
    }

    // Output
    const output = await pipeline[outputFormat]().toBuffer();

    // Speichern
    const url = await uploadToStorage(output, `${imageId}.${outputFormat}`);

    return { url, size: output.length };
  },
  {
    connection,
    concurrency: 2  // CPU-intensive, weniger parallel
  }
);
```

---

## Job Progress & Events

```typescript
// API Route: Job Status
// app/api/jobs/[id]/route.ts
import { emailQueue } from '@/lib/queue';

export async function GET(
  request: Request,
  { params }: { params: { id: string } }
) {
  const job = await emailQueue.getJob(params.id);

  if (!job) {
    return Response.json({ error: 'Job not found' }, { status: 404 });
  }

  const state = await job.getState();
  const progress = job.progress;

  return Response.json({
    id: job.id,
    state,
    progress,
    data: job.data,
    returnValue: job.returnvalue,
    failedReason: job.failedReason,
    timestamp: job.timestamp,
    processedOn: job.processedOn,
    finishedOn: job.finishedOn
  });
}
```

```typescript
// Real-time Updates mit SSE
// app/api/jobs/[id]/stream/route.ts
import { emailQueueEvents } from '@/lib/queue';

export async function GET(
  request: Request,
  { params }: { params: { id: string } }
) {
  const encoder = new TextEncoder();

  const stream = new ReadableStream({
    start(controller) {
      const handleCompleted = ({ jobId, returnvalue }) => {
        if (jobId === params.id) {
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify({
              event: 'completed',
              result: returnvalue
            })}\n\n`)
          );
          controller.close();
        }
      };

      const handleFailed = ({ jobId, failedReason }) => {
        if (jobId === params.id) {
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify({
              event: 'failed',
              error: failedReason
            })}\n\n`)
          );
          controller.close();
        }
      };

      const handleProgress = ({ jobId, data }) => {
        if (jobId === params.id) {
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify({
              event: 'progress',
              progress: data
            })}\n\n`)
          );
        }
      };

      emailQueueEvents.on('completed', handleCompleted);
      emailQueueEvents.on('failed', handleFailed);
      emailQueueEvents.on('progress', handleProgress);

      return () => {
        emailQueueEvents.off('completed', handleCompleted);
        emailQueueEvents.off('failed', handleFailed);
        emailQueueEvents.off('progress', handleProgress);
      };
    }
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive'
    }
  });
}
```

---

## Job Scheduling (Cron)

```typescript
// Wiederkehrende Jobs einrichten
async function setupScheduledJobs() {
  // Täglich um 8:00 UTC
  await reportQueue.add('daily-report', {
    reportType: 'daily'
  }, {
    repeat: {
      pattern: '0 8 * * *'
    },
    jobId: 'daily-report'  // Verhindert Duplikate
  });

  // Wöchentlich Montag 9:00
  await reportQueue.add('weekly-report', {
    reportType: 'weekly'
  }, {
    repeat: {
      pattern: '0 9 * * 1'
    },
    jobId: 'weekly-report'
  });

  // Alle 5 Minuten
  await emailQueue.add('process-email-queue', {}, {
    repeat: {
      every: 5 * 60 * 1000
    }
  });
}

// Scheduled Jobs verwalten
async function listScheduledJobs() {
  const repeatableJobs = await emailQueue.getRepeatableJobs();
  return repeatableJobs;
}

async function removeScheduledJob(key: string) {
  await emailQueue.removeRepeatableByKey(key);
}
```

---

## Dashboard & Monitoring

```typescript
// Bull Board Integration
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

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

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

// In Express/Hono einbinden
app.use('/admin/queues', serverAdapter.getRouter());
```

---

## Best Practices

```typescript
// 1. Idempotente Jobs (können mehrfach ausgeführt werden)
async function processOrder(job: Job<{ orderId: string }>) {
  const order = await db.order.findUnique({
    where: { id: job.data.orderId }
  });

  // Prüfen ob schon verarbeitet
  if (order?.status === 'processed') {
    return { skipped: true, reason: 'Already processed' };
  }

  // Verarbeiten
  await db.order.update({
    where: { id: order.id },
    data: { status: 'processed' }
  });
}

// 2. Graceful Shutdown
process.on('SIGTERM', async () => {
  console.log('Shutting down workers...');
  await emailWorker.close();
  await imageWorker.close();
  process.exit(0);
});

// 3. Dead Letter Queue
const dlQueue = new Queue('dead-letter', { connection });

emailWorker.on('failed', async (job, error) => {
  if (job.attemptsMade >= job.opts.attempts!) {
    await dlQueue.add('failed-email', {
      originalJob: job.data,
      error: error.message,
      failedAt: new Date()
    });
  }
});
```

---

## Fazit

BullMQ bietet:

1. **Zuverlässigkeit**: Retries, Persistence, Acknowledgement
2. **Skalierbarkeit**: Mehrere Worker, Rate Limiting
3. **Flexibilität**: Scheduling, Prioritäten, Progress
4. **Monitoring**: Bull Board, Events, Metrics

Unverzichtbar für Production-Grade Async Processing.

---

## Bildprompts

1. "Queue of tasks being processed by workers, assembly line concept"
2. "Background process running while user continues, async workflow"
3. "Redis connecting multiple workers, distributed job processing"

---

## Quellen

- [BullMQ Documentation](https://docs.bullmq.io/)
- [BullMQ GitHub](https://github.com/taskforcesh/bullmq)
- [Bull Board](https://github.com/felixmosh/bull-board)
- [Redis Pub/Sub](https://redis.io/docs/latest/develop/interact/pubsub/)