import amqp from "amqplib";

export async function init(cb: (key: string, event: string) => boolean | Promise<boolean>) {
  if (!process.env.AMQ_URL || !process.env.AMQ_QUEUE) return;

  const { AMQ_URL: url, AMQ_QUEUE: queue } = process.env;

  const connection = await amqp.connect(url);

  const channel = await connection.createChannel();

  channel.assertQueue(queue, { durable: true });
  channel.prefetch(1);

  console.log("[AMQ] Listening for message...");

  channel.consume(
    queue,
    async (msg) => {
      if (!msg) return;

      const parsed: Record<string, unknown> = JSON.parse(msg.content.toString());

      if (typeof parsed.Key !== "string" || parsed.Key.includes(".keep")) return channel.ack(msg);
      if (typeof parsed.EventName !== "string" || parsed.EventName.includes("Copy")) {
        return channel.ack(msg);
      }

      const key = parsed.Key;
      const event = parsed.EventName;

      if (await cb(key, event)) return channel.ack(msg);

      return await new Promise((resolve) => setTimeout(() => resolve(channel.nack(msg)), 3000));
    },
    { noAck: false },
  );
}

export default { init };
