/** * jobRunner — the single source of truth for enqueueing async jobs. * * Uses InProcessQueue (no Redis) instead of BullMQ. * When a job finishes the worker emits a `job:update` Socket.IO event * to the originating client so the frontend can react in real time. */ import { InProcessQueue } from "./inProcessQueue"; import { io } from "../socket"; import { runEligibilityProcessor } from "./processors/eligibilityProcessor"; import { runClaimStatusProcessor } from "./processors/claimStatusProcessor"; import { runClaimSubmitProcessor } from "./processors/claimSubmitProcessor"; import { runOcrProcessor } from "./processors/ocrProcessor"; import { runDdmaEligibilityProcessor } from "./processors/ddmaEligibilityProcessor"; import { runDeltaInsEligibilityProcessor } from "./processors/deltaInsEligibilityProcessor"; import { runUnitedSCOEligibilityProcessor } from "./processors/unitedSCOEligibilityProcessor"; import { runDentaQuestEligibilityProcessor } from "./processors/dentaQuestEligibilityProcessor"; import { runCCAEligibilityProcessor } from "./processors/ccaEligibilityProcessor"; import { runCCAClaimProcessor } from "./processors/ccaClaimProcessor"; import { runCCAPreAuthProcessor } from "./processors/ccaPreAuthProcessor"; import { runDDMAClaimProcessor } from "./processors/ddmaClaimProcessor"; import { runEligibilityHistoryProcessor } from "./processors/eligibilityHistoryProcessor"; import { runCmspEligibilityHistoryRemainingProcessor } from "./processors/cmspEligibilityHistoryRemainingProcessor"; import type { SeleniumJobData, OcrJobData } from "./queues"; // ── Queue instances ────────────────────────────────────────────────────────── // Selenium: 1 browser at a time (mirrors Python semaphore) const seleniumQ = new InProcessQueue(1); // OCR: allow 2 concurrent (mirrors Python MAX_CONCURRENCY=2) const ocrQ = new InProcessQueue(2); // ── WebSocket helper ───────────────────────────────────────────────────────── function emitJobUpdate( socketId: string | undefined, jobId: string, jobType: string, status: "active" | "completed" | "failed", extra: Record = {} ) { const payload = { jobId, jobType, status, ...extra }; if (socketId && io) { io.to(socketId).emit("job:update", payload); } else if (io) { io.emit("job:update", payload); } } // ── Selenium enqueue ───────────────────────────────────────────────────────── export function enqueueSeleniumJob(data: SeleniumJobData): string { const { jobType, socketId } = data; const jobId = seleniumQ.add(data, async (job) => { emitJobUpdate(socketId, job.id, jobType, "active", { message: "Selenium browser starting…", }); if (jobType === "eligibility-check") { return runEligibilityProcessor({ enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, }); } if (jobType === "claim-status-check") { return runClaimStatusProcessor({ enrichedPayload: data.enrichedPayload, insuranceId: data.insuranceId!, }); } if (jobType === "claim-submit" || jobType === "claim-pre-auth") { return runClaimSubmitProcessor({ enrichedPayload: data.enrichedPayload, files: data.files ?? [], claimId: data.claimId, variant: jobType === "claim-pre-auth" ? "claim-pre-auth" : "claimsubmit", socketId: data.socketId, }); } if (jobType === "ddma-eligibility-check") { return runDdmaEligibilityProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, socketId: data.socketId, }, job.id ); } if (jobType === "deltains-eligibility-check") { return runDeltaInsEligibilityProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, socketId: data.socketId, }, job.id ); } if (jobType === "unitedsco-eligibility-check") { return runUnitedSCOEligibilityProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, socketId: data.socketId, }, job.id ); } if (jobType === "tuftssco-eligibility-check") { return runDentaQuestEligibilityProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, socketId: data.socketId, }, job.id ); } if (jobType === "cca-claim-submit") { return runCCAClaimProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, claimId: data.claimId, socketId: data.socketId, }, job.id ); } if (jobType === "cca-preauth-submit") { return runCCAPreAuthProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, claimId: data.claimId, socketId: data.socketId, }, job.id ); } if (jobType === "ddma-claim-submit") { return runDDMAClaimProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, claimId: data.claimId, socketId: data.socketId, }, job.id ); } if (jobType === "cca-eligibility-check") { return runCCAEligibilityProcessor( { enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, socketId: data.socketId, }, job.id ); } if (jobType === "cmsp-eligibility-history-remaining-check") { return runCmspEligibilityHistoryRemainingProcessor({ enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, }); } if (jobType === "mh-eligibility-history-check") { return runEligibilityHistoryProcessor({ enrichedPayload: data.enrichedPayload, userId: data.userId, insuranceId: data.insuranceId!, formFirstName: data.formFirstName, formLastName: data.formLastName, formDob: data.formDob, }); } throw new Error(`Unknown selenium jobType: ${jobType}`); }); // Attach completion/failure callbacks after the job is in the queue. // We poll the job object once per tick until it settles. (async () => { while (true) { await new Promise((r) => setTimeout(r, 500)); const job = seleniumQ.getJob(jobId); if (!job) break; if (job.status === "completed") { emitJobUpdate(socketId, jobId, jobType, "completed", { result: job.result, }); console.log(`[seleniumQ] job ${jobId} (${jobType}) completed`); break; } if (job.status === "failed") { emitJobUpdate(socketId, jobId, jobType, "failed", { error: job.error, }); console.error(`[seleniumQ] job ${jobId} (${jobType}) failed:`, job.error); break; } } })(); return jobId; } // ── Job status query (HTTP polling fallback) ───────────────────────────────── export function getSeleniumJobStatus(jobId: string) { const job = seleniumQ.getJob(jobId); if (!job) return null; return { status: job.status, result: job.result ?? null, error: job.error ?? null }; } // ── OCR enqueue ────────────────────────────────────────────────────────────── export function enqueueOcrJob(data: OcrJobData): string { const { socketId } = data; const jobId = ocrQ.add(data, async () => { emitJobUpdate(socketId, jobId, "ocr", "active", { message: "OCR processing started…", }); return runOcrProcessor({ files: data.files }); }); (async () => { while (true) { await new Promise((r) => setTimeout(r, 500)); const job = ocrQ.getJob(jobId); if (!job) break; if (job.status === "completed") { emitJobUpdate(socketId, jobId, "ocr", "completed", { result: { rows: job.result }, }); console.log(`[ocrQ] job ${jobId} completed`); break; } if (job.status === "failed") { emitJobUpdate(socketId, jobId, "ocr", "failed", { error: job.error }); console.error(`[ocrQ] job ${jobId} failed:`, job.error); break; } } })(); return jobId; }