//  File:    index.ts
//  Path:    apps/worker/src/index.ts
//  Project: AI Risk Index — airiskindex.io
//  Author:  Simon-Pierre Boucher
//  Contact: contact@spboucher.ai
//  Copyright © 2026 Simon-Pierre Boucher. All rights reserved.
//
//  Description: BullMQ worker: rating batch submit/poll/ingest jobs.

import { Queue, Worker } from "bullmq";
import IORedis from "ioredis";
import { prisma } from "@airiskindex/db";
import { RATER_MODELS, RATER_PROMPT_VERSION, REDIS_URL } from "./config";
import { ingestBatchResults, isBatchComplete, submitRatingBatch, type RatingTask } from "./raters/batch";
import { ratingJobId } from "./raters/job-id";

const connection = new IORedis(REDIS_URL, { maxRetriesPerRequest: null });

export const ratingQueue = new Queue("rating", { connection });

interface SubmitPayload {
  model: string;
}

interface PollPayload {
  model: string;
  batchId: string;
  jobIdToTaskId: Record<string, string>;
}

const worker = new Worker(
  "rating",
  async (job) => {
    if (job.name === "submit") {
      const { model } = job.data as SubmitPayload;
      const tasks = await prisma.task.findMany({
        where: {
          ratings: { none: { model, promptVersion: RATER_PROMPT_VERSION } },
        },
        include: { occupation: { select: { title: true } } },
      });
      if (tasks.length === 0) return { submitted: 0 };

      const ratingTasks: RatingTask[] = tasks.map((task) => ({
        taskId: task.id,
        occupationTitle: task.occupation.title,
        statement: task.statement,
      }));
      const batchId = await submitRatingBatch(ratingTasks, model);
      const jobIdToTaskId = Object.fromEntries(
        ratingTasks.map((task) => [ratingJobId(task.taskId, model, RATER_PROMPT_VERSION), task.taskId]),
      );
      await ratingQueue.add(
        "poll",
        { model, batchId, jobIdToTaskId } satisfies PollPayload,
        { jobId: `poll:${batchId}`, delay: 60_000, attempts: 60, backoff: { type: "fixed", delay: 60_000 } },
      );
      return { submitted: ratingTasks.length, batchId };
    }

    if (job.name === "poll") {
      const { model, batchId, jobIdToTaskId } = job.data as PollPayload;
      if (!(await isBatchComplete(batchId))) {
        throw new Error(`batch ${batchId} still processing`); // retried via backoff
      }
      return ingestBatchResults(batchId, model, new Map(Object.entries(jobIdToTaskId)));
    }

    throw new Error(`unknown job ${job.name}`);
  },
  { connection },
);

worker.on("failed", (job, error) => {
  console.error(`[worker] job ${job?.name}:${job?.id} failed:`, error.message);
});

async function enqueueSubmitJobs(): Promise<void> {
  if (RATER_MODELS.length < 2) {
    console.warn(
      "[worker] RATER_MODELS has fewer than 2 models — multi-model rating is required (CLAUDE.md §6).",
    );
  }
  for (const model of RATER_MODELS) {
    await ratingQueue.add(
      "submit",
      { model } satisfies SubmitPayload,
      { jobId: `submit:${model}:${RATER_PROMPT_VERSION}` },
    );
  }
}

enqueueSubmitJobs().catch((error) => console.error("[worker] enqueue failed:", error));
console.log(`[worker] rating worker up — prompt ${RATER_PROMPT_VERSION}, models: ${RATER_MODELS.join(", ") || "(none)"}`);