feat: add BullMQ queue infrastructure and frontend job status hook

- apps/Backend/src/queue/: connection, queues, workers, processors
- apps/Frontend/src/hooks/use-job-status.ts: WebSocket job progress hook
- apps/Frontend/src/lib/socket.ts: shared Socket.IO singleton

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/Backend/src/queue/processors/_shared.ts

@@ -0,0 +1,156 @@
+/**
+ * Shared utilities used by all job processors.
+ * Avoids duplicating helpers that currently live inside route files.
+ */
+import axios from "axios";
+import PDFDocument from "pdfkit";
+import fsSync from "fs";
+import { storage } from "../../storage";
+import {
+  InsertPatient,
+  insertPatientSchema,
+} from "../../../../../packages/db/types/patient-types";
+
+const SELENIUM_BASE_URL =
+  process.env.SELENIUM_AGENT_BASE_URL || "http://localhost:5002";
+
+// ---------------------------------------------------------------------------
+// Python service helpers
+// ---------------------------------------------------------------------------
+
+/** Start an async job on the Python service and return the session ID. */
+export async function startPythonJob(
+  endpoint: string,
+  payload: any
+): Promise<string> {
+  const resp = await axios.post(
+    `${SELENIUM_BASE_URL}${endpoint}`,
+    payload,
+    { timeout: 10_000 }
+  );
+  const sid: string = resp.data?.session_id;
+  if (!sid) throw new Error(`Python service did not return a session_id from ${endpoint}`);
+  return sid;
+}
+
+/** Poll /job/<sid>/status until completed/failed or timeout. */
+export async function pollPythonJob(
+  sid: string,
+  timeoutMs = 5 * 60 * 1000,
+  intervalMs = 2_000
+): Promise<any> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const resp = await axios.get(
+      `${SELENIUM_BASE_URL}/job/${sid}/status`,
+      { timeout: 5_000 }
+    );
+    const s = resp.data;
+    if (s.status === "completed") {
+      if (s.result?.status === "error") {
+        const msg =
+          typeof s.result.message === "string"
+            ? s.result.message
+            : s.result.message?.msg ?? "Selenium returned error status";
+        throw new Error(msg);
+      }
+      return s.result;
+    }
+    if (s.status === "failed") {
+      throw new Error(s.error || "Python job failed");
+    }
+    await sleep(intervalMs);
+  }
+  throw new Error("Selenium job timed out after polling");
+}
+
+// ---------------------------------------------------------------------------
+// General utilities
+// ---------------------------------------------------------------------------
+
+export function sleep(ms: number) {
+  return new Promise<void>((r) => setTimeout(r, ms));
+}
+
+export function splitName(fullName?: string | null) {
+  if (!fullName) return { firstName: "", lastName: "" };
+  const parts = fullName.trim().split(/\s+/).filter(Boolean);
+  const firstName = parts.shift() ?? "";
+  const lastName = parts.join(" ") ?? "";
+  return { firstName, lastName };
+}
+
+export async function imageToPdfBuffer(imagePath: string): Promise<Buffer> {
+  return new Promise<Buffer>((resolve, reject) => {
+    try {
+      const doc = new PDFDocument({ autoFirstPage: false });
+      const chunks: Uint8Array[] = [];
+      doc.on("data", (c: any) => chunks.push(c));
+      doc.on("end", () => resolve(Buffer.concat(chunks)));
+      doc.on("error", reject);
+
+      const A4_W = 595.28;
+      const A4_H = 841.89;
+      doc.addPage({ size: [A4_W, A4_H] });
+      doc.image(imagePath, 0, 0, { fit: [A4_W, A4_H], align: "center", valign: "center" });
+      doc.end();
+    } catch (e) {
+      reject(e);
+    }
+  });
+}
+
+// ---------------------------------------------------------------------------
+// Patient DB helpers
+// ---------------------------------------------------------------------------
+
+export async function createOrUpdatePatientByInsuranceId(options: {
+  insuranceId: string;
+  firstName?: string | null;
+  lastName?: string | null;
+  dob?: string | Date | null;
+  userId: number;
+}) {
+  const { insuranceId, firstName, lastName, dob, userId } = options;
+  if (!insuranceId) throw new Error("Missing insuranceId");
+
+  const incomingFirst = (firstName || "").trim();
+  const incomingLast = (lastName || "").trim();
+
+  let patient = await storage.getPatientByInsuranceId(insuranceId);
+
+  if (patient && patient.id) {
+    const updates: any = {};
+    if (incomingFirst && String(patient.firstName ?? "").trim() !== incomingFirst)
+      updates.firstName = incomingFirst;
+    if (incomingLast && String(patient.lastName ?? "").trim() !== incomingLast)
+      updates.lastName = incomingLast;
+    if (Object.keys(updates).length > 0) {
+      await storage.updatePatient(patient.id, updates);
+      patient = await storage.getPatientByInsuranceId(insuranceId);
+    }
+    return patient;
+  }
+
+  const createPayload: any = {
+    firstName: incomingFirst,
+    lastName: incomingLast,
+    dateOfBirth: dob,
+    gender: "",
+    phone: "",
+    userId,
+    insuranceId,
+  };
+
+  let patientData: InsertPatient;
+  try {
+    patientData = insertPatientSchema.parse(createPayload);
+  } catch {
+    const safePayload = { ...createPayload };
+    delete safePayload.dateOfBirth;
+    patientData = insertPatientSchema.parse(safePayload);
+  }
+
+  await storage.createPatient(patientData);
+  return storage.getPatientByInsuranceId(insuranceId);
+}

apps/Backend/src/queue/processors/claimStatusProcessor.ts

@@ -0,0 +1,99 @@
+/**
+ * Processor for "claim-status-check" jobs.
+ * Mirrors routes/insuranceStatus.ts /claim-status-check
+ */
+import fs from "fs/promises";
+import fsSync from "fs";
+import path from "path";
+import { storage } from "../../storage";
+import { emptyFolderContainingFile } from "../../utils/emptyTempFolder";
+import {
+  startPythonJob,
+  pollPythonJob,
+  imageToPdfBuffer,
+} from "./_shared";
+
+export interface ClaimStatusProcessorInput {
+  enrichedPayload: any;
+  insuranceId: string; // memberId used to look up the patient
+}
+
+export interface ClaimStatusProcessorResult {
+  pdfUploadStatus?: string;
+  pdfFileId?: number | null;
+}
+
+export async function runClaimStatusProcessor(
+  input: ClaimStatusProcessorInput
+): Promise<ClaimStatusProcessorResult> {
+  const { enrichedPayload, insuranceId } = input;
+
+  // 1) Start async Python job
+  const sid = await startPythonJob("/claim-status-check/async", {
+    data: enrichedPayload,
+  });
+
+  // 2) Poll for completion
+  const result = await pollPythonJob(sid);
+
+  const outputResult: ClaimStatusProcessorResult = {};
+
+  // 3) Look up patient
+  const patient = await storage.getPatientByInsuranceId(insuranceId);
+
+  if (patient && patient.id !== undefined) {
+    let pdfBuffer: Buffer | null = null;
+    let generatedPdfPath: string | null = null;
+
+    if (
+      result.ss_path &&
+      /\.(png|jpg|jpeg)$/i.test(result.ss_path) &&
+      fsSync.existsSync(result.ss_path)
+    ) {
+      try {
+        pdfBuffer = await imageToPdfBuffer(result.ss_path);
+        const pdfFileName = `claimStatus_${insuranceId}_${Date.now()}.pdf`;
+        generatedPdfPath = path.join(path.dirname(result.ss_path), pdfFileName);
+        await fs.writeFile(generatedPdfPath, pdfBuffer);
+      } catch (e) {
+        console.error("[claimStatusProcessor] img→PDF conversion failed:", e);
+        outputResult.pdfUploadStatus = `Failed to convert screenshot to PDF: ${e}`;
+      }
+    } else {
+      outputResult.pdfUploadStatus =
+        "No valid screenshot provided by Selenium; nothing to upload.";
+    }
+
+    if (pdfBuffer && generatedPdfPath) {
+      const groupTitleKey = "CLAIM_STATUS";
+      const groupTitle = "Claim Status";
+
+      let group = await storage.findPdfGroupByPatientTitleKey(patient.id, groupTitleKey);
+      if (!group) group = await storage.createPdfGroup(patient.id, groupTitle, groupTitleKey);
+      if (!group?.id) throw new Error("PDF group creation failed");
+
+      const basename = path.basename(generatedPdfPath);
+      const created = await storage.createPdfFile(group.id, basename, pdfBuffer);
+
+      let createdPdfFileId: number | null = null;
+      if (created && typeof created === "object" && "id" in created) {
+        createdPdfFileId = Number(created.id);
+      }
+
+      outputResult.pdfUploadStatus = `PDF saved to group: ${group.title}`;
+      outputResult.pdfFileId = createdPdfFileId;
+    }
+  } else {
+    outputResult.pdfUploadStatus =
+      "Patient not found; no PDF saved.";
+  }
+
+  // 4) Cleanup
+  try {
+    if (result.ss_path) await emptyFolderContainingFile(result.ss_path);
+  } catch (e) {
+    console.error("[claimStatusProcessor] cleanup failed:", e);
+  }
+
+  return outputResult;
+}

apps/Backend/src/queue/processors/claimSubmitProcessor.ts

@@ -0,0 +1,58 @@
+/**
+ * Processors for "claim-submit" and "claim-pre-auth" jobs.
+ * Mirrors routes/claims.ts /selenium-claim and /selenium-claim-pre-auth
+ */
+import { storage } from "../../storage";
+import { startPythonJob, pollPythonJob } from "./_shared";
+
+export interface ClaimSubmitProcessorInput {
+  enrichedPayload: any;
+  files: { originalname: string; bufferBase64: string; mimetype: string }[];
+  claimId?: number;
+  /** "claimsubmit" (default) or "claim-pre-auth" */
+  variant?: "claimsubmit" | "claim-pre-auth";
+}
+
+export interface ClaimSubmitProcessorResult {
+  status: string;
+  claimNumber?: string;
+  pdf_url?: string;
+  [key: string]: any;
+}
+
+export async function runClaimSubmitProcessor(
+  input: ClaimSubmitProcessorInput
+): Promise<ClaimSubmitProcessorResult> {
+  const { enrichedPayload, files, claimId } = input;
+
+  // Build the same payload shape the Python /claimsubmit endpoint expects
+  const pdfs = files
+    .filter((f) => f.mimetype === "application/pdf")
+    .map(({ originalname, bufferBase64 }) => ({ originalname, bufferBase64 }));
+
+  const images = files
+    .filter((f) => f.mimetype.startsWith("image/"))
+    .map(({ originalname, bufferBase64 }) => ({ originalname, bufferBase64 }));
+
+  const payload = { claim: enrichedPayload, pdfs, images };
+
+  const endpoint =
+    input.variant === "claim-pre-auth" ? "/claim-pre-auth/async" : "/claimsubmit/async";
+
+  // 1) Start async Python job
+  const sid = await startPythonJob(endpoint, payload);
+
+  // 2) Poll for result
+  const result = await pollPythonJob(sid, 10 * 60 * 1000); // claim submit can take up to 10 min
+
+  // 3) Persist claimNumber if returned
+  if (result?.claimNumber && claimId) {
+    try {
+      await storage.updateClaim(claimId, { claimNumber: result.claimNumber });
+    } catch (e) {
+      console.error("[claimSubmitProcessor] failed to persist claimNumber:", e);
+    }
+  }
+
+  return { ...result, claimId };
+}

apps/Backend/src/queue/processors/eligibilityProcessor.ts

@@ -0,0 +1,167 @@
+/**
+ * Processor for "eligibility-check" jobs.
+ *
+ * Replicates the logic from routes/insuranceStatus.ts /eligibility-check
+ * so it can run inside a BullMQ worker without blocking the HTTP server.
+ */
+import fs from "fs/promises";
+import fsSync from "fs";
+import path from "path";
+import { storage } from "../../storage";
+import { emptyFolderContainingFile } from "../../utils/emptyTempFolder";
+import forwardToPatientDataExtractorService from "../../services/patientDataExtractorService";
+import {
+  startPythonJob,
+  pollPythonJob,
+  splitName,
+  createOrUpdatePatientByInsuranceId,
+} from "./_shared";
+
+export interface EligibilityProcessorInput {
+  /** Enriched payload (includes credentials) */
+  enrichedPayload: any;
+  userId: number;
+  insuranceId: string;
+  formFirstName?: string;
+  formLastName?: string;
+  formDob?: string;
+}
+
+export interface EligibilityProcessorResult {
+  patientUpdateStatus?: string;
+  pdfUploadStatus?: string;
+  pdfFileId?: number | null;
+}
+
+export async function runEligibilityProcessor(
+  input: EligibilityProcessorInput
+): Promise<EligibilityProcessorResult> {
+  const {
+    enrichedPayload,
+    userId,
+    insuranceId,
+    formFirstName,
+    formLastName,
+    formDob,
+  } = input;
+
+  // 1) Fire the async Python job
+  const sid = await startPythonJob("/eligibility-check/async", {
+    data: enrichedPayload,
+  });
+
+  // 2) Wait for completion
+  const seleniumResult = await pollPythonJob(sid);
+
+  const outputResult: EligibilityProcessorResult = {};
+
+  // 3) Extract name: prefer selenium extraction → PDF extractor → form input
+  const extracted: { firstName?: string | null; lastName?: string | null } = {};
+
+  if (seleniumResult.firstName || seleniumResult.lastName) {
+    extracted.firstName = seleniumResult.firstName ?? null;
+    extracted.lastName = seleniumResult.lastName ?? null;
+  } else if (seleniumResult.name) {
+    const parts = splitName(seleniumResult.name);
+    extracted.firstName = parts.firstName;
+    extracted.lastName = parts.lastName;
+  }
+
+  if (
+    !extracted.firstName &&
+    !extracted.lastName &&
+    seleniumResult?.pdf_path?.endsWith(".pdf")
+  ) {
+    try {
+      const pdfBuffer = await fs.readFile(seleniumResult.pdf_path);
+      const extraction = await forwardToPatientDataExtractorService({
+        buffer: pdfBuffer,
+        originalname: path.basename(seleniumResult.pdf_path),
+        mimetype: "application/pdf",
+      } as any);
+      if (extraction.name) {
+        const parts = splitName(extraction.name);
+        extracted.firstName = parts.firstName;
+        extracted.lastName = parts.lastName;
+      }
+    } catch (e) {
+      console.error("[eligibilityProcessor] PDF name extraction failed:", e);
+    }
+  }
+
+  const preferFirst = extracted.firstName || formFirstName || null;
+  const preferLast = extracted.lastName || formLastName || null;
+
+  // 4) Create / update patient
+  let patient;
+  try {
+    patient = await createOrUpdatePatientByInsuranceId({
+      insuranceId,
+      firstName: preferFirst,
+      lastName: preferLast,
+      dob: formDob,
+      userId,
+    });
+  } catch (e: any) {
+    throw new Error(`Failed to create/update patient: ${e.message}`);
+  }
+
+  // 5) Update patient status
+  if (patient && patient.id !== undefined) {
+    let newStatus = "UNKNOWN";
+    if (seleniumResult.eligibility === "Y") newStatus = "ACTIVE";
+    else if (seleniumResult.eligibility === "N") newStatus = "INACTIVE";
+
+    const updates: any = { status: newStatus };
+    if (seleniumResult.insurance) updates.insuranceProvider = seleniumResult.insurance;
+
+    await storage.updatePatient(patient.id, updates);
+    outputResult.patientUpdateStatus = `Patient status updated to ${newStatus}`;
+
+    // 6) Save PDF
+    let createdPdfFileId: number | null = null;
+
+    if (seleniumResult.pdf_path?.endsWith(".pdf")) {
+      try {
+        const pdfBuffer = await fs.readFile(seleniumResult.pdf_path);
+        const groupTitleKey = "ELIGIBILITY_STATUS";
+        const groupTitle = "Eligibility Status";
+
+        let group = await storage.findPdfGroupByPatientTitleKey(patient.id, groupTitleKey);
+        if (!group) group = await storage.createPdfGroup(patient.id, groupTitle, groupTitleKey);
+        if (!group?.id) throw new Error("PDF group creation failed");
+
+        const created = await storage.createPdfFile(
+          group.id,
+          path.basename(seleniumResult.pdf_path),
+          pdfBuffer
+        );
+        if (created && typeof created === "object" && "id" in created) {
+          createdPdfFileId = Number(created.id);
+        }
+        outputResult.pdfUploadStatus = `PDF saved to group: ${group.title}`;
+      } catch (e: any) {
+        outputResult.pdfUploadStatus = `PDF upload failed: ${e.message}`;
+      }
+    } else {
+      outputResult.pdfUploadStatus =
+        "No valid PDF path provided by Selenium; nothing uploaded.";
+    }
+
+    outputResult.pdfFileId = createdPdfFileId;
+  } else {
+    outputResult.patientUpdateStatus =
+      "Patient not found or missing ID; no update performed";
+  }
+
+  // 7) Cleanup temp files
+  try {
+    if (seleniumResult.pdf_path) {
+      await emptyFolderContainingFile(seleniumResult.pdf_path);
+    }
+  } catch (e) {
+    console.error("[eligibilityProcessor] cleanup failed:", e);
+  }
+
+  return outputResult;
+}

apps/Backend/src/queue/processors/ocrProcessor.ts

@@ -0,0 +1,42 @@
+/**
+ * Processor for "ocr" jobs.
+ * Calls the PaymentOCR Python service with the uploaded files.
+ */
+import axios from "axios";
+import FormData from "form-data";
+
+const OCR_BASE_URL =
+  process.env.OCR_SERVICE_BASE_URL || "http://localhost:5003";
+
+export interface OcrProcessorInput {
+  files: { originalname: string; bufferBase64: string; mimetype: string }[];
+}
+
+export async function runOcrProcessor(
+  input: OcrProcessorInput
+): Promise<any[]> {
+  const { files } = input;
+
+  const form = new FormData();
+  for (const f of files) {
+    const buf = Buffer.from(f.bufferBase64, "base64");
+    form.append("files", buf, {
+      filename: f.originalname,
+      contentType: f.mimetype,
+      knownLength: buf.length,
+    });
+  }
+
+  const resp = await axios.post<{ rows: any[] }>(
+    `${OCR_BASE_URL}/extract/json`,
+    form,
+    {
+      headers: form.getHeaders(),
+      maxBodyLength: Infinity,
+      maxContentLength: Infinity,
+      timeout: 180_000, // OCR can be heavy; 3-minute limit
+    }
+  );
+
+  return resp.data?.rows ?? [];
+}