import { logError, logInfo } from "@/lib/logger";
import { env } from "@/lib/env";
import amqp from "amqplib";
import type { CampaignService } from "@/v1/services/admin/notifications/campaign.service";
import { getMaxConcurrentCampaigns } from "@/v1/services/admin/notifications/campaign.service";

export class CampaignWorker {
    private service: CampaignService;
    private connection: amqp.ChannelModel | null = null;
    private channel: amqp.Channel | null = null;
    /** Campaign ids this worker has taken off the queue and not yet acked. */
    private readonly inFlight = new Set<string>();

    constructor(service: CampaignService) {
        this.service = service;
    }

    async start() {
        try {
            const connectionString = env.GetString("PRODUCER_QUEUE_CONNECTION");
            const base = env.GetString("NOTIFICATION_QUEUE_NAME") || "campaign_queue";
            const nodeEnv = process.env.NODE_ENV || "development";
            const queue = `${base}_${nodeEnv}`;

            // Own connection, completely independent from the producer singleton
            this.connection = await amqp.connect(connectionString);

            this.connection.on("error", (err) => {
                logError(`[Worker] Connection error: ${err}`);
            });

            this.connection.on("close", () => {
                this.channel = null;
                this.connection = null;
                setTimeout(() => this.start(), 5000);
            });

            this.channel = await this.connection.createChannel();
            logInfo("[Worker] Channel created");

            this.channel.on("error", (err) => {
                logError(`[Worker] Channel error: ${err}`);
            });

            this.channel.on("close", () => {
                logInfo("[Worker] Channel closed");
            });

            const queueInfo = await this.channel.assertQueue(queue, { durable: true });

            // One unacked message per concurrent slot: the worker pulls up to
            // maxConcurrent campaigns and RabbitMQ holds the rest until one finishes.
            const maxConcurrent = getMaxConcurrentCampaigns();
            await this.channel.prefetch(maxConcurrent);
            logInfo(`[Worker] Prefetch set to ${maxConcurrent} — up to ${maxConcurrent} campaigns run in parallel`);

            // The callback is not awaited by amqplib, so several campaigns overlap
            // here by design; each one owns its own message and acks independently.
            const consumerResult = await this.channel.consume(queue, async (msg) => {
                if (msg === null) {
                    logInfo("[Worker] Consumer cancelled by RabbitMQ");
                    return;
                }

                const campaignId = msg.content.toString();
                this.inFlight.add(campaignId);
                logInfo(`[Worker] Campaign ${campaignId} picked up — in flight: ${this.inFlight.size}/${maxConcurrent} [${[...this.inFlight].join(", ")}]`);

                try {
                    const result = await this.service.sendCampaign(campaignId);
                    this.channel?.ack(msg);
                    logInfo(`[Worker] Campaign ${campaignId} finished — processed ${result} recipients`);
                } catch (e) {
                    logError(`[Worker] Campaign ${campaignId} failed: ${e}`);
                    this.channel?.nack(msg, false, false);
                } finally {
                    this.inFlight.delete(campaignId);
                    logInfo(`[Worker] Campaign ${campaignId} released — in flight: ${this.inFlight.size}/${maxConcurrent}`);
                }
            });

        } catch (e) {
            logError(`[Worker] Startup failed: ${e}`);
            setTimeout(() => this.start(), 5000);
        }
    }

    async stop() {
        try {
            if (this.channel) await this.channel.close();
            if (this.connection) await this.connection.close();
            logInfo("[Worker] Stopped.");
        } catch (e) {
            logError(`[Worker] Error stopping: ${e}`);
        }
    }
}