Message Queues with RabbitMQ and SQS in Node.js

const amqp = require('amqplib');

class RabbitMQ {
  constructor() {
    this.connection = null;
    this.channel = null;
  }

  async connect() {
    this.connection = await amqp.connect(process.env.RABBITMQ_URL);
    this.channel = await this.connection.createChannel();

    // Prefetch: process one message at a time per consumer
    await this.channel.prefetch(1);

    this.connection.on('error', (err) => {
      console.error('RabbitMQ connection error:', err);
    });

    this.connection.on('close', () => {
      console.warn('RabbitMQ connection closed, reconnecting...');
      setTimeout(() => this.connect(), 5000);
    });
  }

  async close() {
    await this.channel?.close();
    await this.connection?.close();
  }
}

async function setupExchangeAndQueues(channel) {
  // Declare exchange
  await channel.assertExchange('orders', 'topic', { durable: true });

  // Declare queues with DLX (Dead Letter Exchange)
  await channel.assertExchange('orders.dlx', 'direct', { durable: true });
  await channel.assertQueue('orders.dead-letter', { durable: true });
  await channel.bindQueue('orders.dead-letter', 'orders.dlx', '');

  // Main queue with dead-letter routing
  await channel.assertQueue('orders.process', {
    durable: true,
    arguments: {
      'x-dead-letter-exchange': 'orders.dlx',
      'x-message-ttl': 30000, // 30s timeout
    },
  });

  await channel.bindQueue('orders.process', 'orders', 'order.created');

  // Notification queue
  await channel.assertQueue('orders.notify', { durable: true });
  await channel.bindQueue('orders.notify', 'orders', 'order.*');
}

// Publish a message
async function publishOrder(channel, order) {
  const message = Buffer.from(JSON.stringify({
    id: order.id,
    userId: order.userId,
    total: order.total,
    timestamp: Date.now(),
  }));

  channel.publish('orders', 'order.created', message, {
    persistent: true,         // Survives broker restart
    contentType: 'application/json',
    messageId: `order-${order.id}`,
    headers: {
      'x-retry-count': 0,
    },
  });
}

async function consumeOrders(channel) {
  channel.consume('orders.process', async (msg) => {
    if (!msg) return;

    const retryCount = (msg.properties.headers['x-retry-count'] || 0);
    const order = JSON.parse(msg.content.toString());

    try {
      await processOrder(order);
      channel.ack(msg);
      console.log(`Order ${order.id} processed successfully`);
    } catch (err) {
      console.error(`Order ${order.id} failed (attempt ${retryCount + 1}):`, err.message);

      if (retryCount < 3) {
        // Retry with exponential backoff
        const delay = Math.pow(2, retryCount) * 5000; // 5s, 10s, 20s

        // Republish with incremented retry count
        channel.publish('orders', 'order.created', msg.content, {
          ...msg.properties,
          headers: {
            ...msg.properties.headers,
            'x-retry-count': retryCount + 1,
            'x-delay': delay,
          },
        });
        channel.ack(msg); // Ack original to remove from queue
      } else {
        // Max retries exceeded → dead letter queue
        channel.nack(msg, false, false);
        console.error(`Order ${order.id} sent to DLQ after ${retryCount} retries`);
      }
    }
  });
}

const {
  SQSClient,
  SendMessageCommand,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} = require('@aws-sdk/client-sqs');

const sqs = new SQSClient({ region: 'us-east-1' });
const QUEUE_URL = process.env.SQS_QUEUE_URL;

// Send message
async function sendToSQS(data) {
  await sqs.send(new SendMessageCommand({
    QueueUrl: QUEUE_URL,
    MessageBody: JSON.stringify(data),
    MessageAttributes: {
      EventType: {
        DataType: 'String',
        StringValue: 'order.created',
      },
    },
    // For FIFO queues:
    // MessageGroupId: 'orders',
    // MessageDeduplicationId: `order-${data.id}`,
  }));
}

// Poll for messages
async function pollMessages() {
  while (true) {
    const { Messages } = await sqs.send(new ReceiveMessageCommand({
      QueueUrl: QUEUE_URL,
      MaxNumberOfMessages: 10,
      WaitTimeSeconds: 20,       // Long polling (reduces empty responses)
      VisibilityTimeout: 60,     // 60s to process before reappearing
      MessageAttributeNames: ['All'],
    }));

    if (!Messages?.length) continue;

    await Promise.allSettled(
      Messages.map(async (msg) => {
        try {
          const data = JSON.parse(msg.Body);
          await processOrder(data);

          // Delete after successful processing
          await sqs.send(new DeleteMessageCommand({
            QueueUrl: QUEUE_URL,
            ReceiptHandle: msg.ReceiptHandle,
          }));
        } catch (err) {
          console.error('Failed to process message:', err);
          // Message will reappear after VisibilityTimeout
        }
      })
    );
  }
}

async function processOrder(order) {
  // Check if already processed (using a unique idempotency key)
  const existing = await db.processedEvents.findOne({
    eventId: `order-${order.id}`,
  });

  if (existing) {
    console.log(`Order ${order.id} already processed, skipping`);
    return;
  }

  // Process the order
  await db.transaction(async (tx) => {
    await tx.orders.updateStatus(order.id, 'processing');
    await tx.inventory.reserve(order.items);
    await tx.processedEvents.create({
      eventId: `order-${order.id}`,
      processedAt: new Date(),
    });
  });
}