H1-H10 优化OCR实时体验闭环

2026-05-09 11:48:57 +08:00
@@ -1,6 +1,7 @@
 const cron = require("node-cron");

 const { buildMissingRequiredOcrError, runOcrForInvoice } = require("../services/ocr");
+const { publishInvoiceUpdated } = require("../services/events");
 const { resolveInvoiceDedupBlocker } = require("../services/invoiceGovernance");
 const { readFileBuffer } = require("../services/storage");

@@ -70,7 +71,30 @@ async function claimPendingInvoices(prisma, config, workerId) {
        ocr_lock_owner = ${workerId}
    FROM candidates
    WHERE i.id = candidates.id
-    RETURNING i.id, i.owner_employee_id, i.file_sha256, i.ocr_provider, i.ocr_type, i.storage_path, i.mime_type, i.ocr_attempts
+    RETURNING i.id, i.owner_employee_id, i.uploaded_by_employee_id, i.file_sha256, i.ocr_provider, i.ocr_type, i.storage_path, i.mime_type, i.ocr_attempts
+  `;
+}
+
+async function claimPendingInvoiceById(prisma, config, workerId, invoiceId) {
+  const lockTtlSeconds = Math.max(60, Number(config.ocrLockTtlSeconds || 300));
+
+  return prisma.$queryRaw`
+    WITH candidates AS (
+      SELECT id
+      FROM "invoice_pool"
+      WHERE id = ${invoiceId}::uuid
+        AND status = 'pending_ocr'
+        AND (ocr_locked_at IS NULL OR ocr_locked_at < NOW() - make_interval(secs => ${lockTtlSeconds}))
+      ORDER BY created_at ASC
+      LIMIT 1
+      FOR UPDATE SKIP LOCKED
+    )
+    UPDATE "invoice_pool" AS i
+    SET ocr_locked_at = NOW(),
+        ocr_lock_owner = ${workerId}
+    FROM candidates
+    WHERE i.id = candidates.id
+    RETURNING i.id, i.owner_employee_id, i.uploaded_by_employee_id, i.file_sha256, i.ocr_provider, i.ocr_type, i.storage_path, i.mime_type, i.ocr_attempts
  `;
 }

@@ -226,11 +250,12 @@ async function markInvoiceAsFailed(prisma, claimedInvoice, errorMessage, config)
  const now = new Date();
  const nextAttempts = Number(claimedInvoice.ocr_attempts || 0) + 1;
  const isTerminal = nextAttempts >= config.ocrMaxAttempts;
+  const nextStatus = isTerminal ? "ocr_failed" : "pending_ocr";

  await prisma.invoicePool.update({
    where: { id: claimedInvoice.id },
    data: {
-      status: isTerminal ? "ocr_failed" : "pending_ocr",
+      status: nextStatus,
      ocr_provider: getEffectiveOcrProvider(config),
      ocr_attempts: nextAttempts,
      ocr_last_error: errorMessage,
@@ -239,6 +264,8 @@ async function markInvoiceAsFailed(prisma, claimedInvoice, errorMessage, config)
      ocr_lock_owner: null,
    },
  });
+
+  return nextStatus;
 }

 async function processClaimedInvoice(prisma, config, claimedInvoice, workerId, logger) {
@@ -276,7 +303,7 @@ async function processClaimedInvoice(prisma, config, claimedInvoice, workerId, l
        },
        "OCR invoice marked as failed for missing required fields",
      );
-      return { outcome: "failed" };
+      return { outcome: "failed", status: "ocr_failed" };
    }

    const duplicate = await resolveInvoiceDedupBlocker(
@@ -302,14 +329,14 @@ async function processClaimedInvoice(prisma, config, claimedInvoice, workerId, l
        },
        "OCR invoice marked as duplicate",
      );
-      return { outcome: "duplicate" };
+      return { outcome: "duplicate", status: "ocr_failed" };
    }

    await markInvoiceAsAvailable(prisma, config, claimedInvoice.id, normalized, ocrResultId);
-    return { outcome: "available" };
+    return { outcome: "available", status: "available" };
  } catch (error) {
    const message = trimErrorMessage(error);
-    await markInvoiceAsFailed(prisma, claimedInvoice, message, config);
+    const status = await markInvoiceAsFailed(prisma, claimedInvoice, message, config);
    logger.error(
      {
        err: error,
@@ -320,12 +347,11 @@ async function processClaimedInvoice(prisma, config, claimedInvoice, workerId, l
      },
      "OCR processing failed",
    );
-    return { outcome: "failed" };
+    return { outcome: "failed", status };
  }
 }

-async function runOcrOnce(prisma, config, logger, workerId = `ocr:${process.pid}`) {
-  const claimed = await claimPendingInvoices(prisma, config, workerId);
+async function processClaimedInvoices(prisma, config, logger, workerId, claimed, options = {}) {
  if (!claimed.length) {
    return { claimed: 0, available: 0, duplicate: 0, failed: 0 };
  }
@@ -341,11 +367,31 @@ async function runOcrOnce(prisma, config, logger, workerId = `ocr:${process.pid}
    } else if (result?.outcome === "failed") {
      stats.failed += 1;
    }
+
+    if (typeof options.onInvoiceProcessed === "function") {
+      await options.onInvoiceProcessed({
+        invoiceId: item.id,
+        status: result?.status ?? item.status ?? null,
+        outcome: result?.outcome ?? null,
+        ownerEmployeeId: item.owner_employee_id ?? null,
+        uploadedByEmployeeId: item.uploaded_by_employee_id ?? null,
+      });
+    }
  }

  return stats;
 }

+async function runOcrOnce(prisma, config, logger, workerId = `ocr:${process.pid}`, options = {}) {
+  const claimed = await claimPendingInvoices(prisma, config, workerId);
+  return processClaimedInvoices(prisma, config, logger, workerId, claimed, options);
+}
+
+async function runOcrForInvoiceId(prisma, config, logger, invoiceId, workerId = `ocr:${process.pid}`, options = {}) {
+  const claimed = await claimPendingInvoiceById(prisma, config, workerId, invoiceId);
+  return processClaimedInvoices(prisma, config, logger, workerId, claimed, options);
+}
+
 function startOcrJob(fastify) {
  if (!["tencent", "mock"].includes(fastify.config.ocrProvider)) {
    fastify.log.info({ provider: fastify.config.ocrProvider }, "OCR job disabled");
@@ -359,6 +405,12 @@ function startOcrJob(fastify) {

  let running = false;
  const workerId = `ocr-worker:${process.pid}`;
+  const onInvoiceProcessed = (event) =>
+    publishInvoiceUpdated(fastify, {
+      invoiceId: event.invoiceId,
+      status: event.status,
+      audienceEmployeeIds: [event.ownerEmployeeId, event.uploadedByEmployeeId],
+    });

  const task = cron.schedule(fastify.config.ocrPollCron, async () => {
    if (running) {
@@ -367,7 +419,9 @@ function startOcrJob(fastify) {

    running = true;
    try {
-      const result = await runOcrOnce(fastify.prisma, fastify.config, fastify.log, workerId);
+      const result = await runOcrOnce(fastify.prisma, fastify.config, fastify.log, workerId, {
+        onInvoiceProcessed,
+      });
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: "ocr.tick",
        driver: "cron",
@@ -423,9 +477,11 @@ function startOcrJob(fastify) {

 module.exports = {
  OCR_DUPLICATE_ERROR,
+  claimPendingInvoiceById,
  claimPendingInvoices,
  getEffectiveOcrProvider,
  markInvoiceAsQualityFailed,
+  runOcrForInvoiceId,
  runOcrOnce,
  startOcrJob,
 };
@@ -2,16 +2,18 @@ const PgBoss = require("pg-boss");

 const { revokeExpiredDelegations } = require("./delegationRevokeJob");
 const { archiveUnusedInvoices } = require("./invoiceArchiveJob");
-const { runOcrOnce } = require("./ocrJob");
+const { runOcrForInvoiceId, runOcrOnce } = require("./ocrJob");
 const { runVerificationForRecord } = require("./verificationJob");
 const { runIntegrationOutboxOnce } = require("./integrationOutboxJob");
 const { runErpMasterDataSyncOnce } = require("./erpMasterDataSyncJob");
+const { publishInvoiceUpdated } = require("../services/events");
 const {
  FEATURE_FLAG_KEYS,
  evaluateFeatureFlag,
 } = require("../services/featureFlags");

 const JOB_OCR_TICK = "ocr.tick";
+const JOB_OCR_INVOICE = "ocr.invoice";
 const JOB_DELEGATION_REVOKE = "delegation.revoke-expired";
 const JOB_INVOICE_ARCHIVE = "invoice.archive-unused";
 const JOB_INVOICE_VERIFY = "invoice.verify";
@@ -43,6 +45,7 @@ async function startPgBossJobs(fastify) {
  });

  await boss.createQueue(JOB_OCR_TICK);
+  await boss.createQueue(JOB_OCR_INVOICE);
  await boss.createQueue(JOB_DELEGATION_REVOKE);
  await boss.createQueue(JOB_INVOICE_ARCHIVE);
  await boss.createQueue(JOB_INVOICE_VERIFY);
@@ -51,6 +54,13 @@ async function startPgBossJobs(fastify) {
    await boss.createQueue(JOB_ERP_MASTER_DATA_SYNC);
  }

+  const onOcrInvoiceProcessed = (event) =>
+    publishInvoiceUpdated(fastify, {
+      invoiceId: event.invoiceId,
+      status: event.status,
+      audienceEmployeeIds: [event.ownerEmployeeId, event.uploadedByEmployeeId],
+    });
+
  await boss.work(JOB_DELEGATION_REVOKE, { batchSize: 1 }, async (jobs) => {
    const job = jobs?.[0];
    try {
@@ -122,7 +132,9 @@ async function startPgBossJobs(fastify) {
    const job = jobs?.[0];
    const workerId = `pgboss:ocr:${process.pid}`;
    try {
-      const stats = await runOcrOnce(fastify.prisma, fastify.config, fastify.log, workerId);
+      const stats = await runOcrOnce(fastify.prisma, fastify.config, fastify.log, workerId, {
+        onInvoiceProcessed: onOcrInvoiceProcessed,
+      });
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_OCR_TICK,
        driver: "pgboss",
@@ -164,6 +176,67 @@ async function startPgBossJobs(fastify) {
    }
  });

+  await boss.work(JOB_OCR_INVOICE, { batchSize: 1 }, async (jobs) => {
+    if (!["tencent", "mock"].includes(fastify.config.ocrProvider)) {
+      return;
+    }
+
+    const job = jobs?.[0];
+    const invoiceId = job?.data?.invoiceId;
+    if (!invoiceId) {
+      return;
+    }
+
+    const workerId = `pgboss:ocr-invoice:${process.pid}`;
+    try {
+      const stats = await runOcrForInvoiceId(
+        fastify.prisma,
+        fastify.config,
+        fastify.log,
+        invoiceId,
+        workerId,
+        { onInvoiceProcessed: onOcrInvoiceProcessed },
+      );
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: JOB_OCR_INVOICE,
+        driver: "pgboss",
+        outcome: "success",
+      });
+      if (stats.available > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          { job: JOB_OCR_INVOICE, driver: "pgboss", kind: "available" },
+          stats.available,
+        );
+      }
+      if (stats.duplicate > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          { job: JOB_OCR_INVOICE, driver: "pgboss", kind: "duplicate" },
+          stats.duplicate,
+        );
+      }
+      if (stats.failed > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          { job: JOB_OCR_INVOICE, driver: "pgboss", kind: "failed" },
+          stats.failed,
+        );
+      }
+      if (stats.claimed > 0) {
+        fastify.log.info(
+          { jobId: job?.id, invoiceId, workerId, ...stats },
+          "pg-boss OCR invoice job processed",
+        );
+      }
+    } catch (error) {
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: JOB_OCR_INVOICE,
+        driver: "pgboss",
+        outcome: "failure",
+      });
+      fastify.log.error({ err: error, jobId: job?.id, invoiceId }, "pg-boss OCR invoice job failed");
+      throw error;
+    }
+  });
+
  await boss.work(JOB_INVOICE_VERIFY, { batchSize: 1 }, async (jobs) => {
    const job = jobs?.[0];
    const recordId = job?.data?.recordId;
@@ -301,7 +374,7 @@ async function startPgBossJobs(fastify) {
    await boss.schedule(JOB_OCR_TICK, fastify.config.ocrPollCron, null, {
      tz: fastify.config.appTimezone,
      singletonKey: JOB_OCR_TICK,
-      singletonSeconds: 55,
+      singletonSeconds: 1,
    });
  } else {
    await boss.unschedule(JOB_OCR_TICK).catch(() => {});
@@ -359,12 +432,27 @@ async function startPgBossJobs(fastify) {
  return boss;
 }

+async function enqueueOcrInvoice(fastify, invoiceId) {
+  if (
+    fastify?.config?.jobDriver === "pgboss" &&
+    fastify.boss &&
+    ["tencent", "mock"].includes(fastify.config.ocrProvider)
+  ) {
+    await fastify.boss.send(JOB_OCR_INVOICE, { invoiceId });
+    return { mode: "queued" };
+  }
+
+  return { mode: "fallback" };
+}
+
 module.exports = {
+  JOB_OCR_INVOICE,
  JOB_OCR_TICK,
  JOB_DELEGATION_REVOKE,
  JOB_INVOICE_ARCHIVE,
  JOB_INVOICE_VERIFY,
  JOB_INTEGRATION_OUTBOX,
  JOB_ERP_MASTER_DATA_SYNC,
+  enqueueOcrInvoice,
  startPgBossJobs,
 };