spb/airiskindex Public

1//  File:    index.ts2//  Path:    apps/worker/src/index.ts3//  Project: AI Risk Index — airiskindex.io4//  Author:  Simon-Pierre Boucher5//  Contact: contact@spboucher.ai6//  Copyright © 2026 Simon-Pierre Boucher. All rights reserved.7//8//  Description: BullMQ worker: rating batch submit/poll/ingest jobs.910import { Queue, Worker } from "bullmq";11import IORedis from "ioredis";12import { prisma } from "@airiskindex/db";13import { RATER_MODELS, RATER_PROMPT_VERSION, REDIS_URL } from "./config";14import { ingestBatchResults, isBatchComplete, submitRatingBatch, type RatingTask } from "./raters/batch";15import { ratingJobId } from "./raters/job-id";1617const connection = new IORedis(REDIS_URL, { maxRetriesPerRequest: null });1819export const ratingQueue = new Queue("rating", { connection });2021interface SubmitPayload {22  model: string;23}2425interface PollPayload {26  model: string;27  batchId: string;28  jobIdToTaskId: Record<string, string>;29}3031const worker = new Worker(32  "rating",33  async (job) => {34    if (job.name === "submit") {35      const { model } = job.data as SubmitPayload;36      const tasks = await prisma.task.findMany({37        where: {38          ratings: { none: { model, promptVersion: RATER_PROMPT_VERSION } },39        },40        include: { occupation: { select: { title: true } } },41      });42      if (tasks.length === 0) return { submitted: 0 };4344      const ratingTasks: RatingTask[] = tasks.map((task) => ({45        taskId: task.id,46        occupationTitle: task.occupation.title,47        statement: task.statement,48      }));49      const batchId = await submitRatingBatch(ratingTasks, model);50      const jobIdToTaskId = Object.fromEntries(51        ratingTasks.map((task) => [ratingJobId(task.taskId, model, RATER_PROMPT_VERSION), task.taskId]),52      );53      await ratingQueue.add(54        "poll",55        { model, batchId, jobIdToTaskId } satisfies PollPayload,56        { jobId: `poll:${batchId}`, delay: 60_000, attempts: 60, backoff: { type: "fixed", delay: 60_000 } },57      );58      return { submitted: ratingTasks.length, batchId };59    }6061    if (job.name === "poll") {62      const { model, batchId, jobIdToTaskId } = job.data as PollPayload;63      if (!(await isBatchComplete(batchId))) {64        throw new Error(`batch ${batchId} still processing`); // retried via backoff65      }66      return ingestBatchResults(batchId, model, new Map(Object.entries(jobIdToTaskId)));67    }6869    throw new Error(`unknown job ${job.name}`);70  },71  { connection },72);7374worker.on("failed", (job, error) => {75  console.error(`[worker] job ${job?.name}:${job?.id} failed:`, error.message);76});7778async function enqueueSubmitJobs(): Promise<void> {79  if (RATER_MODELS.length < 2) {80    console.warn(81      "[worker] RATER_MODELS has fewer than 2 models — multi-model rating is required (CLAUDE.md §6).",82    );83  }84  for (const model of RATER_MODELS) {85    await ratingQueue.add(86      "submit",87      { model } satisfies SubmitPayload,88      { jobId: `submit:${model}:${RATER_PROMPT_VERSION}` },89    );90  }91}9293enqueueSubmitJobs().catch((error) => console.error("[worker] enqueue failed:", error));94console.log(`[worker] rating worker up — prompt ${RATER_PROMPT_VERSION}, models: ${RATER_MODELS.join(", ") || "(none)"}`);95