import { Worker, WorkerOptions, Processor } from "bullmq";

import { getQueue, QueueName } from "@app/lib/queue/QueueRegistry";

import { mailProcessor } from "@app/lib/queue/processors/mail";

import { insightProcessor } from "@app/lib/queue/processors/insights";

import { InsightJobName } from "@app/lib/queue/types";

type ProcessorMap = Record<QueueName, Processor>;

const processorMap: ProcessorMap = {

  mail: mailProcessor,

  insights: insightProcessor,

};

// Per-queue concurrency

// TODO: When we process images, we need to bump for that queue

const concurrencyMap: Record<QueueName, number> = {

  mail: 10,

  insights: 5,

};

const workers: Worker[] = [];

export function bootWorkers(): void {

  console.log("Booting workers...");

  for (const [name, processor] of Object.entries(processorMap)) {

    const queueName = name as QueueName;

    console.log(`Worker ${queueName} booting...`);

    const worker = new Worker(queueName, processor, {

      connection: {

        host: process.env.VALKEY_HOST,

        port: Number(process.env.VALKEY_PORT),

      },

      concurrency: concurrencyMap[queueName],

    } satisfies WorkerOptions);

    worker.on("failed", (job, err) => {

      console.error(`[${queueName}] job ${job?.id} failed:`, err.message);

    });

    workers.push(worker);

  }

}

export async function scheduleInsights(): Promise<void> {

  const queue = getQueue("insights");

  await queue.add(

    InsightJobName.FanOut,

    {},

    {

      repeat: { pattern: "0 3 * * 1" }, // Mondays at 3am

      jobId: "insight-fan-out-weekly", // prevents duplicate repeatable jobs on restart

    },

  );

}

export async function closeAllWorkers(): Promise<void> {

  await Promise.all(workers.map((w) => w.close()));

}