// This message will ALWAYS fail - it's a poison pill
const poisonPill = {
  orderId: 'ORD-999',
  items: [{ productId: null, quantity: -3 }], // Invalid data
  userId: undefined // Missing required field
};

// Consumer crashes every time it tries to process this
queue.subscribe('orders', async (message) => {
  const { orderId, items, userId } = message.data;

  // TypeError: Cannot read property 'name' of undefined
  const user = await db.users.findById(userId);
  const email = user.email; // CRASH - user is null because userId was undefined

  // This code never runs. Message gets redelivered. Crashes again.
});

// Consumer with retry limit
queue.subscribe('orders', {
  maxRetries: 3,
  onMaxRetries: 'deadLetterQueue' // Send to DLQ after 3 failures
}, async (message) => {
  try {
    await processOrder(message.data);
    message.ack();
  } catch (err) {
    console.error(`Attempt ${message.retryCount + 1} failed:`, err.message);
    message.nack(); // Will be retried up to 3 times, then DLQ
  }
});

Main Queue
   |
   | (processing fails)
   v
Retry Queue (delay: 5 seconds)
   |
   | (fails again)
   v
Retry Queue (delay: 30 seconds)
   |
   | (fails again)
   v
Retry Queue (delay: 5 minutes)
   |
   | (fails again)
   v
Dead Letter Queue (manual investigation required)

// Retry topology with exponential backoff
class RetryableConsumer {
  constructor(mainQueue, dlq, options = {}) {
    this.mainQueue = mainQueue;
    this.dlq = dlq;
    this.maxRetries = options.maxRetries || 5;
    this.baseDelay = options.baseDelay || 1000; // 1 second
  }

  async processWithRetry(message, handler) {
    const retryCount = message.headers['x-retry-count'] || 0;

    try {
      await handler(message.data);
      message.ack();
    } catch (err) {
      if (retryCount >= this.maxRetries) {
        // Max retries exceeded: send to DLQ
        console.error(
          `Message ${message.id} failed ${retryCount} times, sending to DLQ`
        );
        await this.dlq.publish({
          originalMessage: message.data,
          error: err.message,
          failedAt: new Date().toISOString(),
          retryCount,
          queue: this.mainQueue.name
        });
        message.ack(); // Remove from main queue
        return;
      }

      // Calculate delay with exponential backoff + jitter
      const delay = this.baseDelay * Math.pow(2, retryCount);
      const jitter = Math.random() * delay * 0.1; // 10% jitter
      const totalDelay = delay + jitter;

      console.log(
        `Retry ${retryCount + 1}/${this.maxRetries} in ${totalDelay}ms`
      );

      // Republish with incremented retry count and delay
      await this.mainQueue.publish(message.data, {
        headers: { 'x-retry-count': retryCount + 1 },
        delay: totalDelay
      });
      message.ack(); // Remove the current copy
    }
  }
}

// Usage
const consumer = new RetryableConsumer(ordersQueue, ordersDLQ, {
  maxRetries: 5,
  baseDelay: 2000 // 2s, 4s, 8s, 16s, 32s then DLQ
});

ordersQueue.subscribe(async (message) => {
  await consumer.processWithRetry(message, processOrder);
});

// What a DLQ message should contain
const dlqMessage = {
  // The original message, untouched
  originalMessage: {
    orderId: 'ORD-456',
    items: [{ productId: 'PROD-789', quantity: 2 }],
    userId: 'USER-123'
  },

  // Metadata about the failure
  metadata: {
    sourceQueue: 'orders.process',
    firstFailedAt: '2025-03-15T10:30:00Z',
    lastFailedAt: '2025-03-15T10:35:12Z',
    retryCount: 5,
    lastError: 'PaymentGatewayTimeout: Connection timed out after 30s',
    errorStack: 'Error: PaymentGatewayTimeout\n    at charge (payment.js:42)',
    consumerVersion: '2.3.1',
    environment: 'production'
  }
};

// Manual replay tool
class DLQReplayTool {
  async listMessages(filters = {}) {
    const messages = await this.dlq.peek(100); // Look without consuming
    return messages.filter(msg => {
      if (filters.sourceQueue && msg.metadata.sourceQueue !== filters.sourceQueue) {
        return false;
      }
      if (filters.errorType && !msg.metadata.lastError.includes(filters.errorType)) {
        return false;
      }
      return true;
    });
  }

  async replayOne(messageId) {
    const msg = await this.dlq.getMessage(messageId);
    await this.originalQueue.publish(msg.originalMessage);
    await this.dlq.delete(messageId);
    console.log(`Replayed message ${messageId}`);
  }

  async replayAll(filter) {
    const messages = await this.listMessages(filter);
    console.log(`Replaying ${messages.length} messages...`);

    for (const msg of messages) {
      await this.replayOne(msg.id);
      await sleep(100); // Don't overwhelm the main queue
    }
  }
}

// DLQ monitoring configuration
const dlqAlerts = {
  // Alert immediately: any message in a critical DLQ
  critical: {
    queues: ['payments-dlq', 'orders-dlq'],
    threshold: 1,
    action: 'page-oncall',
    message: 'Critical DLQ has messages. Data is not being processed.'
  },

  // Alert after 10 messages: non-critical DLQ growing
  warning: {
    queues: ['emails-dlq', 'analytics-dlq'],
    threshold: 10,
    action: 'slack-channel',
    message: 'DLQ depth growing. Check consumer health.'
  },

  // Alert on rate: sudden spike in DLQ messages
  rateAlert: {
    window: '5m',
    threshold: 50, // More than 50 DLQ messages in 5 minutes
    action: 'page-oncall',
    message: 'DLQ spike detected. Possible upstream issue or deployment bug.'
  }
};