fix: 增加回单核对与付款登记门禁（09-08-reimbursement-payment-closeout）

server/src/jobs/jobSystem.js

@@ -1,6 +1,7 @@
 const { startDelegationRevokeJob } = require("./delegationRevokeJob");
 const { startInvoiceArchiveJob } = require("./invoiceArchiveJob");
 const { startOcrJob } = require("./ocrJob");
+const { startReceiptOcrJob } = require("./paymentReceiptOcrJob");
 const { startIntegrationOutboxWorker } = require("./integrationOutboxWorker");
 const { startErpMasterDataSyncJob } = require("./erpMasterDataSyncJob");
 const { startPgBossJobs } = require("./pgBossJobs");
@@ -18,6 +19,7 @@ async function startJobs(fastify) {
 
   startDelegationRevokeJob(fastify);
   startOcrJob(fastify);
+  startReceiptOcrJob(fastify);
   startInvoiceArchiveJob(fastify);
   startIntegrationOutboxWorker(fastify);
   startErpMasterDataSyncJob(fastify);

server/src/jobs/paymentReceiptOcrJob.js

@@ -0,0 +1,135 @@
+const { createHash } = require("node:crypto");
+const { readFileBuffer } = require("../services/storage");
+const { recognizePaymentReceipt } = require("../services/paymentReceiptOcr");
+const { storeReceiptRecognition } = require("../services/paymentReceiptVerification");
+const { requiresBankReceipt } = require("../services/paymentProcessingProgress");
+const { resolveFieldEncryptionKey } = require("../services/crypto");
+
+async function recordReceiptSystemFailure(prisma, attachment, state) {
+  try {
+    return await prisma.$transaction(async (tx) => {
+      await tx.$queryRaw`SELECT id FROM payment_batches WHERE id = ${attachment.payment_batch_id}::uuid FOR UPDATE`;
+      const batch = await tx.paymentBatch.findUnique({ where: { id: attachment.payment_batch_id } });
+      if (!batch || !["reviewed", "failed"].includes(batch.status)) return { status: "discarded" };
+      const result = await tx.paymentBatchReceiptAttachment.updateMany({
+        where: { id: attachment.id, payment_batch_id: attachment.payment_batch_id,
+          receipt_ocr_version: attachment.receipt_ocr_version, file_sha256: attachment.file_sha256,
+          receipt_ocr_state: "processing", archived_at: null, governance_status: "approved" },
+        data: { receipt_ocr_state: state },
+      });
+      return { status: result.count === 1 ? state : "discarded" };
+    }, { maxWait: 5000, timeout: 10000 });
+  } catch {
+    return { status: "failure_recording_failed" };
+  }
+}
+
+async function runReceiptOcrTick(fastify) {
+  try {
+    const result = await runReceiptOcrOnce({ prisma: fastify.prisma, config: fastify.config });
+    if (!["idle", "matched", "needs_review", "discarded"].includes(result.status)) {
+      fastify.log.warn({ outcome: result.status }, "Receipt OCR requires attention");
+    }
+    return result;
+  } catch {
+    fastify.log.warn({ outcome: "tick_failed" }, "Receipt OCR tick failed");
+    return { status: "tick_failed" };
+  }
+}
+
+function startReceiptOcrJob(fastify) {
+  const cron = require("node-cron");
+  let running = false;
+  const task = cron.schedule(fastify.config.ocrPollCron, async () => {
+    if (running) return;
+    running = true;
+    try { await runReceiptOcrTick(fastify); }
+    finally { running = false; }
+  });
+  fastify.addHook("onClose", async () => { task.stop(); task.destroy(); });
+  return task;
+}
+
+async function startReceiptPgBossJob(fastify, boss) {
+  const queue = "payment.receipt.ocr";
+  await boss.createQueue(queue, { retryLimit: 0, expireInSeconds: 900 });
+  await boss.work(queue, { batchSize: 1 }, async () => runReceiptOcrTick(fastify));
+  await boss.schedule(queue, fastify.config.ocrPollCron, null, {
+    tz: fastify.config.appTimezone, singletonKey: queue, retryLimit: 0,
+  });
+}
+
+async function claimReceiptJob(prisma) {
+  return prisma.$transaction(async (tx) => {
+    const batches = await tx.$queryRaw`
+      SELECT batch.id FROM payment_batches batch
+      WHERE batch.status IN ('reviewed', 'failed') AND EXISTS (
+        SELECT 1 FROM payment_batch_receipt_attachments attachment
+        WHERE attachment.payment_batch_id = batch.id
+          AND attachment.receipt_ocr_state = 'queued'
+          AND attachment.archived_at IS NULL AND attachment.governance_status = 'approved'
+      ) ORDER BY batch.id FOR UPDATE SKIP LOCKED LIMIT 1
+    `;
+    if (!batches.length) return null;
+    const batch = await tx.paymentBatch.findUnique({ where: { id: batches[0].id }, include: { items: true } });
+    if (batch?.entity_type !== "reimbursement" || !requiresBankReceipt(batch)) {
+      await tx.paymentBatchReceiptAttachment.updateMany({
+        where: { payment_batch_id: batches[0].id, receipt_ocr_state: "queued" },
+        data: { receipt_ocr_state: "legacy" },
+      });
+      return null;
+    }
+    const attachments = await tx.$queryRaw`
+      SELECT id FROM payment_batch_receipt_attachments
+      WHERE payment_batch_id = ${batches[0].id}::uuid AND receipt_ocr_state = 'queued'
+        AND archived_at IS NULL AND governance_status = 'approved'
+      ORDER BY created_at, id FOR UPDATE SKIP LOCKED LIMIT 1
+    `;
+    if (!attachments.length) return null;
+    return tx.paymentBatchReceiptAttachment.update({
+      where: { id: attachments[0].id },
+      data: { receipt_ocr_state: "processing", receipt_ocr_started_at: new Date(),
+        receipt_ocr_version: { increment: 1 } },
+    });
+  }, { timeout: 10000 });
+}
+
+async function runReceiptOcrOnce({ prisma, config, readBuffer = readFileBuffer, recognize = recognizePaymentReceipt }) {
+  const attachment = await claimReceiptJob(prisma);
+  if (!attachment) return { status: "idle" };
+  if (!resolveFieldEncryptionKey(config)) {
+    return recordReceiptSystemFailure(prisma, attachment, "key_failed");
+  }
+  let recognition;
+  try {
+    if (attachment.file_size <= 0 || attachment.file_size > 5 * 1024 * 1024) throw new Error("size");
+    const buffer = await readBuffer({ config, storagePath: attachment.storage_path });
+    if (buffer.length !== attachment.file_size ||
+        createHash("sha256").update(buffer).digest("hex") !== attachment.file_sha256) throw new Error("integrity");
+    recognition = await recognize({ buffer, mimeType: attachment.mime_type, config });
+  } catch {
+    recognition = { status: "ocr_failed", pagesExpected: 0, pagesProcessed: 0, pages: [],
+      reason: "receipt_processing_failed" };
+  }
+  try {
+    return await prisma.$transaction(async (tx) => {
+      const verification = await storeReceiptRecognition(tx, config, {
+        batchId: attachment.payment_batch_id, attachmentId: attachment.id,
+        sourceSha256: attachment.file_sha256, expectedVersion: attachment.receipt_ocr_version, recognition,
+      });
+      await tx.paymentBatchReceiptAttachment.update({
+        where: { id: attachment.id },
+        data: { receipt_ocr_state: recognition.status === "ocr_failed" ? "ocr_failed" : verification.status },
+      });
+      return { status: recognition.status === "ocr_failed" ? "ocr_failed" : verification.status };
+    }, { timeout: 10000 });
+  } catch (error) {
+    if (["receipt_job_superseded", "receipt_source_changed", "receipt_unavailable", "receipt_batch_not_editable"]
+      .includes(error.code)) return { status: "discarded" };
+    const state = ["receipt_key_unavailable", "receipt_key_rotation_required"].includes(error.code) ? "key_failed" :
+      ["receipt_bank_unsupported", "receipt_business_unsupported"].includes(error.code) ? "snapshot_failed" : "system_failed";
+    return recordReceiptSystemFailure(prisma, attachment, state);
+  }
+}
+
+module.exports = { claimReceiptJob, runReceiptOcrOnce, runReceiptOcrTick, startReceiptOcrJob, startReceiptPgBossJob };

server/src/jobs/pgBossJobs.js

@@ -1,4 +1,5 @@
 const PgBoss = require("pg-boss");
+const { startReceiptPgBossJob } = require("./paymentReceiptOcrJob");
 
 const { revokeExpiredDelegations } = require("./delegationRevokeJob");
 const { archiveUnusedInvoices } = require("./invoiceArchiveJob");
@@ -45,6 +46,7 @@ async function startPgBossJobs(fastify) {
   });
 
   await boss.createQueue(JOB_OCR_TICK);
+  await startReceiptPgBossJob(fastify, boss);
   await boss.createQueue(JOB_OCR_INVOICE);
   await boss.createQueue(JOB_DELEGATION_REVOKE);
   await boss.createQueue(JOB_INVOICE_ARCHIVE);