// ❌ Synchronous — user waits for everything
app.post('/register', async (req, res) => {
  const user = await createUser(req.body);
  await sendWelcomeEmail(user.email);      // 500ms
  await updateAnalytics(user);              // 200ms
  await notifyAdmin(user);                  // 300ms
  res.json({ success: true });             // User waited 1 second!
});

// ✅ Async — instant response, tasks queued
app.post('/register', async (req, res) => {
  const user = await createUser(req.body);
  await emailQueue.add({ email: user.email });     // ~5ms
  await analyticsQueue.add({ userId: user.id });   // ~5ms
  await notificationQueue.add({ userId: user.id }); // ~5ms
  res.json({ success: true });                     // 15ms total!
});

Synchronous (Blocking):                    Async with Queue:

Client         Server                     Client         Server         Queue         Worker
  │              │                          │              │              │              │
  ├── POST ─────►│                          ├── POST ─────►│              │              │
  │              ├── Email ───────►[slow]   │◄── 202 ──────┤              │              │
  │              ├── SMS ─────────►[slow]   │              ├── Add Job ───►│              │
  │              ├── Analytics ───►[slow]   │              │              ├── Pick Up ──►│
  │◄── 200 ──────┤                          │              │              │              ├── Email
  │              │                          │              │              │              ├── SMS
                                            │              │              │              ├── Analytics
                                            │              │              │              │
                                            │              │              │◄── Done ──────┤
              User waited 3+ seconds                      User responded in 15ms!

flowchart TD
    subgraph Producers["📤 Producers"]
        API["Express API"]
        Cron["Scheduled Jobs"]
        Events["Event Handlers"]
    end

    subgraph BullMQ["BullMQ (Redis)"]
        Queue["Queue<br/>(Redis Lists/Sorted Sets)"]
        SubQueue["Sub-Queue System"]
        Wait["⏳ Waiting Jobs"]
        Active["⚡ Active Jobs"]
        Delay["⏰ Delayed Jobs"]
        Fail["❌ Failed Jobs"]
        Complete["✅ Completed Jobs"]
    end

    subgraph Workers["⚙️ Workers"]
        W1["Worker Process 1"]
        W2["Worker Process 2"]
        W3["Worker Process N"]
    end

    API --> Queue
    Cron --> Queue
    Events --> Queue

    Queue --> Wait
    Wait --> Active
    Delay --> Wait
    Active --> Fail
    Active --> Complete
    Active --> W1
    Active --> W2
    Active --> W3

    style Queue fill:#4f46e5,color:#fff
    style Wait fill:#7c3aed,color:#fff
    style Active fill:#059669,color:#fff
    style Fail fill:#dc2626,color:#fff
    style Complete fill:#10b981,color:#fff

stateDiagram-v2
    [*] --> Waiting: queue.add()
    Waiting --> Active: Worker picks up
    Active --> Completed: job.moveToCompleted()
    Active --> Failed: job.moveToFailed()
    Active --> Waiting: Retry (with delay)
    Failed --> Waiting: Manual retry
    Failed --> DeadLetter: Max retries exceeded

    state Completed {
        [*] --> Cleaned: removeOnComplete
    }

    state Failed {
        [*] --> DeadLetter: No more retries
    }

flowchart TD
    subgraph Clients["📱 Clients"]
        Web["Web App"]
        Mobile["Mobile App"]
    end

    subgraph Gateway["API Gateway"]
        API["Express Server"]
    end

    subgraph Queues["📨 Message Queues"]
        OrderQ["Order Queue"]
        EmailQ["Email Queue"]
        AnalyticsQ["Analytics Queue"]
        NotifQ["Notification Queue"]
    end

    subgraph Services["⚙️ Microservices"]
        OrderS["Order Service"]
        EmailS["Email Service"]
        AnalyticsS["Analytics Service"]
        NotifS["Notification Service"]
    end

    subgraph Data["🗄️ Data Stores"]
        DB["PostgreSQL"]
        Cache["Redis"]
        Search["Elasticsearch"]
    end

    Web --> API
    Mobile --> API
    API --> OrderQ
    OrderQ --> OrderS
    OrderQ --> EmailQ
    OrderQ --> AnalyticsQ
    OrderS --> EmailQ
    EmailS --> NotifQ
    AnalyticsS --> Search
    OrderS --> DB
    OrderS --> Cache
    EmailS --> DB

sequenceDiagram
    participant U as User
    participant API as API Server
    participant Q as BullMQ Queue
    participant OS as Order Worker
    participant PS as Payment Worker
    participant ES as Email Worker
    participant DB as Database

    U->>API: POST /orders (checkout)
    API->>DB: Create order (status: pending)
    API->>Q: order.process { orderId: 123 }
    API-->>U: 202 { orderId: 123, status: "processing" }

    OS->>Q: Pick up job
    OS->>Q: Update progress (10%)
    OS->>DB: Update order (status: confirmed)
    OS->>Q: order.payment { orderId: 123, amount: 49.99 }

    PS->>Q: Pick up payment job
    PS->>DB: Charge payment
    PS->>Q: order.email-confirmation { orderId: 123 }

    ES->>Q: Pick up email job
    ES->>DB: Send email receipt
    ES->>Q: Mark job completed

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis(process.env.REDIS_URL);

// Define a queue
const emailQueue = new Queue('email', { connection });

// Add a job
await emailQueue.add('welcome-email', {
  to: 'user@example.com',
  name: 'Alice',
  template: 'welcome',
}, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
  removeOnComplete: { count: 100 },
});

// Process jobs
const worker = new Worker('email', async (job) => {
  const { to, name } = job.data;
  await sendEmail({ to, subject: `Welcome ${name}!` });
  return { sent: true };
}, { connection });

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');
const express = require('express');

const app = express();
app.use(express.json());

const connection = new Redis(process.env.REDIS_URL);
const emailQueue = new Queue('email', { connection });

// Express endpoint — instant response
app.post('/register', async (req, res) => {
  const user = await User.create(req.body);

  // Queue the email — don't block the response
  await emailQueue.add('welcome', {
    email: user.email,
    name: user.name,
    userId: user.id,
  }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 5000 },
  });

  res.status(201).json({
    message: 'User created. Welcome email will be sent shortly.',
    userId: user.id,
  });
});

// Worker (can be in a separate process)
const worker = new Worker('email', async (job) => {
  console.log(`Sending ${job.name} to ${job.data.email}`);

  switch (job.name) {
    case 'welcome':
      await sendTemplateEmail(job.data.email, 'welcome', { name: job.data.name });
      break;
    case 'reset-password':
      await sendTemplateEmail(job.data.email, 'reset-password', { token: job.data.token });
      break;
  }

  return { sent: true, to: job.data.email };
}, { connection, concurrency: 5 });

worker.on('completed', (job) => {
  console.log(`✅ Job ${job.id} completed`);
});

worker.on('failed', (job, err) => {
  console.error(`❌ Job ${job.id} failed after ${job.attemptsMade} attempts:`, err.message);
});

app.listen(3000);

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');
const connection = new Redis(process.env.REDIS_URL);

const orderQueue = new Queue('order-processing', { connection });

// Order processing pipeline
app.post('/orders', async (req, res) => {
  const order = await Order.create(req.body);

  // Step 1: Validate payment
  const paymentJob = await orderQueue.add('validate-payment', {
    orderId: order.id,
    amount: order.total,
    cardToken: req.body.cardToken,
  }, { attempts: 2 });

  // Step 2: After payment succeeds, reserve inventory
  const inventoryJob = await orderQueue.add('reserve-inventory', {
    orderId: order.id,
    items: order.items,
  }, {
    dependsOn: [paymentJob.id], // Only after payment succeeds!
    attempts: 3,
  });

  // Step 3: After inventory reserved, send confirmation
  await orderQueue.add('send-confirmation', {
    orderId: order.id,
    email: req.body.email,
  }, {
    dependsOn: [inventoryJob.id],
    delay: 5000, // Send confirmation 5 seconds after inventory reserved
  });

  // Step 4: Schedule a follow-up email 24 hours later
  await orderQueue.add('follow-up-email', {
    orderId: order.id,
    email: req.body.email,
  }, {
    delay: 24 * 60 * 60 * 1000, // 24 hours
    attempts: 3,
  });

  res.status(202).json({ orderId: order.id });
});

// Worker processes
const worker = new Worker('order-processing', async (job) => {
  switch (job.name) {
    case 'validate-payment':
      return await processPayment(job.data);
    case 'reserve-inventory':
      return await reserveStock(job.data);
    case 'send-confirmation':
      return await sendOrderEmail(job.data);
    case 'follow-up-email':
      return await sendFollowUp(job.data);
  }
}, { connection });

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');
const axios = require('axios');
const crypto = require('crypto');

const connection = new Redis(process.env.REDIS_URL);
const webhookQueue = new Queue('webhooks', { connection });

// Webhook delivery function
async function deliverWebhook(url, payload, secret) {
  const timestamp = Date.now();
  const signature = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${JSON.stringify(payload)}`)
    .digest('hex');

  const response = await axios.post(url, payload, {
    headers: {
      'Content-Type': 'application/json',
      'X-Webhook-Signature': signature,
      'X-Webhook-Timestamp': timestamp,
    },
    timeout: 10000,
    validateStatus: () => true, // Don't throw on non-2xx
  });

  // Classify the response
  if (response.status >= 200 && response.status < 300) {
    return { status: 'success', statusCode: response.status };
  }

  // 4xx errors (client errors) — don't retry, the endpoint is misconfigured
  if (response.status >= 400 && response.status < 500) {
    throw new NonRetriableError(`Webhook rejected: ${response.status} ${response.data}`);
  }

  // 5xx errors — retry with exponential backoff
  throw new Error(`Server error: ${response.status}`);
}

class NonRetriableError extends Error {
  constructor(message) {
    super(message);
    this.name = 'NonRetriableError';
  }
}

// Add webhook job
async function sendWebhook(event, url, payload, secret) {
  await webhookQueue.add(event, {
    url, payload, secret,
    event,
  }, {
    attempts: 5,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { age: 3600 * 24 }, // Keep for 24 hours
    removeOnFail: { age: 3600 * 24 * 7 }, // Keep failed for 7 days
  });
}

// Webhook worker
const worker = new Worker('webhooks', async (job) => {
  const { url, payload, secret } = job.data;

  // Check if the endpoint is rate-limited
  const backoff = await getEndpointBackoff(url);
  if (backoff > 0) {
    throw new Error(`Endpoint rate-limited. Retry in ${backoff}ms`);
  }

  return await deliverWebhook(url, payload, secret);
}, { connection });

worker.on('failed', async (job, err) => {
  if (err instanceof NonRetriableError) {
    // Don't retry — client error
    await job.discard();
    await notifyAdmin(`Webhook permanently failed: ${job.data.url}`, err.message);
  } else if (job.attemptsMade >= job.opts.attempts) {
    // All retries exhausted
    await notifyAdmin(`Webhook exhausted retries: ${job.data.url}`, err.message);
  }
});

// Scheduled cleanup for completed jobs
const { QueueScheduler } = require('bullmq');
new QueueScheduler('webhooks', { connection });

// ❌ Don't store sensitive data in jobs
await queue.add('send-email', { password: 'plain-text-password' });

// ✅ Store references, look up sensitive data in the worker
await queue.add('send-email', { userId: 123 });
// In worker: const user = await User.findById(job.data.userId);

// ✅ Production worker configuration
const worker = new Worker('queue', async (job) => {
  try {
    const result = await processJob(job);
    return result;
  } catch (err) {
    // Log with structured logging
    logger.error({ jobId: job.id, jobName: job.name, err }, 'Job failed');
    throw err; // Re-throw for BullMQ retry mechanism
  }
}, {
  connection,
  concurrency: 5,
  lockDuration: 30000,     // Job lock timeout (30s)
  stalledInterval: 30000,  // Check for stalled jobs every 30s
  maxStalledCount: 3,      // Max times a job can be stalled
});

worker.on('completed', (job) => {
  logger.info({ jobId: job.id }, 'Job completed');
});

worker.on('failed', (job, err) => {
  // Alert if retries exhausted
  if (job.attemptsMade >= job.opts.attempts) {
    alertDevops(`Job permanently failed: ${job.id}`, err);
  }
});

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');
const connection = new Redis();

const queue = new Queue('tasks', { connection });

// Bug 1: What happens if the processor throws but there's no error handler?
const worker = new Worker('tasks', async (job) => {
  if (job.data.type === 'fail') {
    throw new Error('Processing failed');
  }
  console.log('Processed:', job.id);
}, { connection });

// Bug 2: No 'failed' event handler — we won't know jobs are failing!

// Bug 3: No retry configuration — one failure and the job is dead forever
await queue.add('task', { type: 'slow' });
// Bug 4: What if the slow task times out without lockDuration configured?

// Bug 5: Worker runs in the same process as the API — blocking issue?
app.get('/fast', (req, res) => {
  res.json({ ok: true });
  // This hangs if worker is busy!
});

// Quick reference: BullMQ

// 1. Setup
const { Queue, Worker } = require('bullmq');
const connection = new (require('ioredis'))(process.env.REDIS_URL);

// 2. Queue
const queue = new Queue('name', { connection });

// 3. Add job
await queue.add('job-name', { data: 'value' }, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
  delay: 5000, // 5 second delay
  jobId: 'unique-id', // Deduplication
});

// 4. Worker
const worker = new Worker('name', async (job) => {
  // Process job
  return { result: 'success' };
}, { connection, concurrency: 5 });

// 5. Events
worker.on('completed', (job) => {});
worker.on('failed', (job, err) => {});
worker.on('progress', (job, progress) => {});

// 6. Queue metrics
await queue.getWaitingCount();
await queue.getActiveCount();
await queue.getCompletedCount();
await queue.getFailedCount();