Message Queues

import amqp from "amqplib";

// --- Configuration ---
const RABBITMQ_URL = process.env.RABBITMQ_URL || "amqp://localhost:5672";
const EXCHANGE = "app.events";
const QUEUE = "email.notifications";
const DLQ = "email.notifications.dlq";
const MAX_RETRIES = 3;

// --- Types ---
interface EmailMessage {
  to: string;
  subject: string;
  template: string;
  data: Record<string, unknown>;
  userId: string;
  timestamp: string;
}

// --- Producer ---
class MessageProducer {
  private connection: amqp.Connection | null = null;
  private channel: amqp.Channel | null = null;

  async connect(): Promise<void> {
    this.connection = await amqp.connect(RABBITMQ_URL);
    this.channel = await this.connection.createConfirmChannel();

    // Declare exchange (topic type for routing key matching)
    await this.channel.assertExchange(EXCHANGE, "topic", { durable: true });

    // Declare dead letter queue
    await this.channel.assertQueue(DLQ, {
      durable: true,
      arguments: { "x-message-ttl": 7 * 24 * 60 * 60 * 1000 }, // 7 day retention
    });

    // Declare main queue with DLQ binding
    await this.channel.assertQueue(QUEUE, {
      durable: true,
      arguments: {
        "x-dead-letter-exchange": "",
        "x-dead-letter-routing-key": DLQ,
      },
    });

    // Bind queue to exchange with routing key
    await this.channel.bindQueue(QUEUE, EXCHANGE, "user.#");

    console.log("Producer connected to RabbitMQ");
  }

  async publish(routingKey: string, message: EmailMessage): Promise<boolean> {
    if (!this.channel) throw new Error("Not connected");

    const payload = Buffer.from(JSON.stringify(message));

    return new Promise((resolve, reject) => {
      this.channel!.publish(
        EXCHANGE,
        routingKey,
        payload,
        {
          persistent: true,               // survive broker restart
          contentType: "application/json",
          messageId: crypto.randomUUID(),
          timestamp: Date.now(),
          headers: { "x-retry-count": 0 },
        },
        (err) => {
          if (err) {
            console.error("Publish failed:", err);
            reject(err);
          } else {
            resolve(true);
          }
        }
      );
    });
  }

  async close(): Promise<void> {
    await this.channel?.close();
    await this.connection?.close();
  }
}

// --- Consumer ---
class MessageConsumer {
  private connection: amqp.Connection | null = null;
  private channel: amqp.Channel | null = null;

  async connect(): Promise<void> {
    this.connection = await amqp.connect(RABBITMQ_URL);
    this.channel = await this.connection.createChannel();

    // Process 5 messages at a time (prevents one slow consumer from getting all messages)
    await this.channel.prefetch(5);

    console.log("Consumer connected to RabbitMQ");
  }

  async consume(
    handler: (msg: EmailMessage) => Promise<void>
  ): Promise<void> {
    if (!this.channel) throw new Error("Not connected");

    await this.channel.consume(
      QUEUE,
      async (msg) => {
        if (!msg) return;

        const retryCount = (msg.properties.headers?.["x-retry-count"] as number) || 0;

        try {
          const payload: EmailMessage = JSON.parse(msg.content.toString());

          console.log(`Processing: ${payload.subject} (attempt ${retryCount + 1})`);

          // Process the message
          await handler(payload);

          // Acknowledge success
          this.channel!.ack(msg);
          console.log(`Processed: ${payload.subject}`);
        } catch (err) {
          console.error(`Failed to process message:`, err);

          if (retryCount < MAX_RETRIES) {
            // Retry: reject and requeue with incremented retry count
            this.channel!.nack(msg, false, false); // don't requeue

            // Re-publish with incremented retry count and delay
            const delay = Math.pow(2, retryCount) * 1000; // exponential backoff
            setTimeout(() => {
              this.channel!.publish("", QUEUE, msg.content, {
                ...msg.properties,
                headers: {
                  ...msg.properties.headers,
                  "x-retry-count": retryCount + 1,
                },
              });
            }, delay);
          } else {
            // Max retries exceeded — send to DLQ
            console.error(
              `Message ${msg.properties.messageId} sent to DLQ after ${MAX_RETRIES} retries`
            );
            this.channel!.nack(msg, false, false); // goes to DLQ via exchange config
          }
        }
      },
      { noAck: false }
    );
  }

  async close(): Promise<void> {
    await this.channel?.close();
    await this.connection?.close();
  }
}

// --- Email processing logic ---
async function sendEmail(msg: EmailMessage): Promise<void> {
  // Real implementation would use SendGrid, SES, etc.
  console.log(`Sending email to ${msg.to}: ${msg.subject}`);

  // Simulate occasional failures for demonstration
  if (Math.random() < 0.1) {
    throw new Error("SMTP connection failed");
  }

  // Simulate processing time
  await new Promise((resolve) => setTimeout(resolve, 100));
}

// --- Main ---
async function runProducer(): Promise<void> {
  const producer = new MessageProducer();
  await producer.connect();

  // Publish some messages
  await producer.publish("user.welcome", {
    to: "user@example.com",
    subject: "Welcome to the platform",
    template: "welcome",
    data: { name: "John" },
    userId: "user-123",
    timestamp: new Date().toISOString(),
  });

  await producer.publish("user.password-reset", {
    to: "user@example.com",
    subject: "Password Reset",
    template: "password-reset",
    data: { resetLink: "https://app.com/reset?token=abc" },
    userId: "user-123",
    timestamp: new Date().toISOString(),
  });

  console.log("Messages published");
  await producer.close();
}

async function runConsumer(): Promise<void> {
  const consumer = new MessageConsumer();
  await consumer.connect();
  await consumer.consume(sendEmail);
  console.log("Consumer waiting for messages...");

  process.on("SIGTERM", async () => {
    console.log("Shutting down consumer...");
    await consumer.close();
    process.exit(0);
  });
}

// Run based on CLI arg
const mode = process.argv[2];
if (mode === "producer") {
  runProducer().catch(console.error);
} else {
  runConsumer().catch(console.error);
}