jobs-monorepo/services/scraper-glassdoor/utils/rabbitmq.ts

import * as amqp from "amqplib";

export interface RabbitMQConfig {
  url: string;
  queueName: string;
  exchangeName: string;
  exchangeType: "direct" | "topic" | "fanout" | "headers";
  durable: boolean;
}

export interface SearchQuery {
  keywords: string;
  location: string;
  fwt?: string;
  numPages: number;
}

export class RabbitMQClient {
  private connection: amqp.ChannelModel | null = null;
  private channel: amqp.Channel | null = null;
  private config: RabbitMQConfig;

  constructor(config: RabbitMQConfig) {
    this.config = config;
  }

  async connect(): Promise<void> {
    try {
      console.log(`Connecting to RabbitMQ at: ${this.config.url}`);
      this.connection = await amqp.connect(this.config.url);
      console.log("Connected to RabbitMQ successfully");

      if (!this.connection) {
        throw new Error("Failed to establish RabbitMQ connection");
      }

      this.channel = await this.connection.createChannel();

      if (!this.channel) {
        throw new Error("Failed to create RabbitMQ channel");
      }

      // Ensure exchange and queue exist
      await this.channel.assertExchange(
        this.config.exchangeName,
        this.config.exchangeType,
        { durable: this.config.durable }
      );

      await this.channel.assertQueue(this.config.queueName, {
        durable: this.config.durable,
        arguments: {
          "x-dead-letter-exchange": "scraper_dlx",
          "x-dead-letter-routing-key": this.config.queueName,
          "x-message-ttl": 24 * 60 * 60 * 1000, // 24 hours in milliseconds
          "x-max-retries": 3,
        },
      });

      await this.channel.bindQueue(
        this.config.queueName,
        this.config.exchangeName,
        this.config.queueName
      );

      // Set prefetch to process one message at a time
      await this.channel.prefetch(1);

      // Handle connection events
      this.connection.on("close", () => {
        console.log("RabbitMQ connection closed");
        this.connection = null;
        this.channel = null;
      });

      this.connection.on("error", (error: Error) => {
        console.error("RabbitMQ connection error:", error);
        this.connection = null;
        this.channel = null;
      });
    } catch (error) {
      console.error("Failed to connect to RabbitMQ:", error);
      throw error;
    }
  }

  async subscribe(
    messageHandler: (message: SearchQuery) => Promise<void>
  ): Promise<void> {
    if (!this.channel) {
      throw new Error(
        "RabbitMQ channel not initialized. Call connect() first."
      );
    }

    console.log(
      `Waiting for messages from ${this.config.queueName}. To exit press CTRL+C`
    );

    await this.channel.consume(
      this.config.queueName,
      async (msg: amqp.ConsumeMessage | null) => {
        if (msg && this.channel) {
          try {
            const messageContent = msg.content.toString();
            console.log(`Received message: ${messageContent}`);

            const searchQuery: SearchQuery = JSON.parse(messageContent);
            console.log(`Processing Glassdoor scrape request:`, searchQuery);

            // Process the message using the provided handler
            await messageHandler(searchQuery);

            // Acknowledge the message on success
            this.channel.ack(msg);
            console.log("Message processed successfully");
          } catch (error) {
            console.error("Error processing message:", error);
            // Reject the message and don't requeue it
            if (this.channel) {
              this.channel.nack(msg, false, false);
            }
          }
        }
      }
    );
  }

  async close(): Promise<void> {
    try {
      if (this.channel) {
        await this.channel.close();
        this.channel = null;
      }
      if (this.connection) {
        await this.connection.close();
        this.connection = null;
      }
      console.log("RabbitMQ connection closed gracefully");
    } catch (error) {
      console.error("Error closing RabbitMQ connection:", error);
    }
  }

  isConnected(): boolean {
    return this.connection !== null && this.channel !== null;
  }
}

export function createRabbitMQConfig(): RabbitMQConfig {
  return {
    url: process.env.RABBITMQ_URL || "amqp://guest:guest@localhost:5672/",
    queueName: process.env.GLASSDOOR_QUEUE_NAME || "scraper.glassdoor",
    exchangeName: process.env.SCRAPER_EXCHANGE_NAME || "scraper_exchange",
    exchangeType: "topic",
    durable: true,
  };
}

export async function createRabbitMQClient(): Promise<RabbitMQClient> {
  const config = createRabbitMQConfig();
  const client = new RabbitMQClient(config);
  await client.connect();
  return client;
}