Wellspringmanagement01/apps/Backend/src/queue/jobRunner.ts

/**
 * 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 type { SeleniumJobData, OcrJobData } from "./queues";

// ── Queue instances ──────────────────────────────────────────────────────────
// Selenium: 1 browser at a time (mirrors Python semaphore)
const seleniumQ = new InProcessQueue<SeleniumJobData>(1);
// OCR: allow 2 concurrent (mirrors Python MAX_CONCURRENCY=2)
const ocrQ = new InProcessQueue<OcrJobData>(2);

// ── WebSocket helper ─────────────────────────────────────────────────────────
function emitJobUpdate(
  socketId: string | undefined,
  jobId: string,
  jobType: string,
  status: "active" | "completed" | "failed",
  extra: Record<string, any> = {}
) {
  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",
      });
    }
    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-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
      );
    }
    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;
}

// ── 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;
}