feat: 完成P6-P8后端能力与P9路由重构（V1-V2,D1,D3,B1-B2,B4,I1-I4,P9A2-P9A3）

2026-02-10 11:28:28 +08:00
@@ -0,0 +1,41 @@
+const cron = require("node-cron");
+
+const { runIntegrationOutboxOnce } = require("./integrationOutboxJob");
+
+function startIntegrationOutboxWorker(fastify) {
+  let running = false;
+
+  const task = cron.schedule(fastify.config.integrationOutboxCron, async () => {
+    if (running) {
+      return;
+    }
+
+    running = true;
+    try {
+      await runIntegrationOutboxOnce(fastify.prisma, fastify.config, fastify.log);
+    } catch (error) {
+      fastify.log.error({ err: error }, "Integration outbox worker tick failed");
+    } finally {
+      running = false;
+    }
+  });
+
+  fastify.addHook("onClose", async () => {
+    task.stop();
+    task.destroy();
+  });
+
+  fastify.log.info(
+    {
+      cron: fastify.config.integrationOutboxCron,
+      batchSize: fastify.config.integrationOutboxBatchSize,
+    },
+    "Integration outbox worker scheduled",
+  );
+
+  return task;
+}
+
+module.exports = {
+  startIntegrationOutboxWorker,
+};
@@ -1,6 +1,7 @@
 const { startDelegationRevokeJob } = require("./delegationRevokeJob");
 const { startInvoiceArchiveJob } = require("./invoiceArchiveJob");
 const { startOcrJob } = require("./ocrJob");
+const { startIntegrationOutboxWorker } = require("./integrationOutboxWorker");
 const { startPgBossJobs } = require("./pgBossJobs");

 async function startJobs(fastify) {
@@ -17,10 +18,10 @@ async function startJobs(fastify) {
  startDelegationRevokeJob(fastify);
  startOcrJob(fastify);
  startInvoiceArchiveJob(fastify);
+  startIntegrationOutboxWorker(fastify);
  return { driver: "cron" };
 }

 module.exports = {
  startJobs,
 };
-
@@ -3,10 +3,14 @@ const PgBoss = require("pg-boss");
 const { revokeExpiredDelegations } = require("./delegationRevokeJob");
 const { archiveUnusedInvoices } = require("./invoiceArchiveJob");
 const { runOcrOnce } = require("./ocrJob");
+const { runVerificationForRecord } = require("./verificationJob");
+const { runIntegrationOutboxOnce } = require("./integrationOutboxJob");

 const JOB_OCR_TICK = "ocr.tick";
 const JOB_DELEGATION_REVOKE = "delegation.revoke-expired";
 const JOB_INVOICE_ARCHIVE = "invoice.archive-unused";
+const JOB_INVOICE_VERIFY = "invoice.verify";
+const JOB_INTEGRATION_OUTBOX = "integration.outbox.dispatch";

 function resolveDatabaseConnectionString() {
  const connectionString = process.env.DATABASE_URL;
@@ -35,6 +39,8 @@ async function startPgBossJobs(fastify) {
  await boss.createQueue(JOB_OCR_TICK);
  await boss.createQueue(JOB_DELEGATION_REVOKE);
  await boss.createQueue(JOB_INVOICE_ARCHIVE);
+  await boss.createQueue(JOB_INVOICE_VERIFY);
+  await boss.createQueue(JOB_INTEGRATION_OUTBOX);

  await boss.work(JOB_DELEGATION_REVOKE, { batchSize: 1 }, async (jobs) => {
    const job = jobs?.[0];
@@ -149,6 +155,96 @@ async function startPgBossJobs(fastify) {
    }
  });

+  await boss.work(JOB_INVOICE_VERIFY, { batchSize: 1 }, async (jobs) => {
+    const job = jobs?.[0];
+    const recordId = job?.data?.recordId;
+    if (!recordId) {
+      return;
+    }
+
+    try {
+      const result = await runVerificationForRecord(
+        fastify.prisma,
+        fastify.config,
+        recordId,
+      );
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: JOB_INVOICE_VERIFY,
+        driver: "pgboss",
+        outcome: "success",
+      });
+      if (!result?.skipped) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          {
+            job: JOB_INVOICE_VERIFY,
+            driver: "pgboss",
+            kind: result.status ?? "unknown",
+          },
+          1,
+        );
+      }
+    } catch (error) {
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: JOB_INVOICE_VERIFY,
+        driver: "pgboss",
+        outcome: "failure",
+      });
+      fastify.log.error({ err: error, jobId: job?.id }, "pg-boss invoice verify failed");
+      throw error;
+    }
+  });
+
+  await boss.work(JOB_INTEGRATION_OUTBOX, { batchSize: 1 }, async (jobs) => {
+    const job = jobs?.[0];
+
+    try {
+      const result = await runIntegrationOutboxOnce(
+        fastify.prisma,
+        fastify.config,
+        fastify.log,
+      );
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: JOB_INTEGRATION_OUTBOX,
+        driver: "pgboss",
+        outcome: "success",
+      });
+      if (result.succeeded > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          {
+            job: JOB_INTEGRATION_OUTBOX,
+            driver: "pgboss",
+            kind: "succeeded",
+          },
+          result.succeeded,
+        );
+      }
+      if (result.failed > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          {
+            job: JOB_INTEGRATION_OUTBOX,
+            driver: "pgboss",
+            kind: "failed",
+          },
+          result.failed,
+        );
+      }
+      if (result.total > 0) {
+        fastify.log.info(
+          { jobId: job?.id, ...result },
+          "pg-boss integration outbox tick processed",
+        );
+      }
+    } catch (error) {
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: JOB_INTEGRATION_OUTBOX,
+        driver: "pgboss",
+        outcome: "failure",
+      });
+      fastify.log.error({ err: error, jobId: job?.id }, "pg-boss integration outbox failed");
+      throw error;
+    }
+  });
+
  await boss.schedule(JOB_DELEGATION_REVOKE, fastify.config.delegationRevokeCron, null, {
    tz: fastify.config.appTimezone,
    singletonKey: JOB_DELEGATION_REVOKE,
@@ -171,13 +267,32 @@ async function startPgBossJobs(fastify) {
    await boss.unschedule(JOB_OCR_TICK).catch(() => {});
  }

+  if (fastify.config.invoiceVerificationEnabled) {
+    await boss.schedule(JOB_INVOICE_VERIFY, fastify.config.verificationPollCron, null, {
+      tz: fastify.config.appTimezone,
+      singletonKey: JOB_INVOICE_VERIFY,
+      singletonSeconds: 55,
+    });
+  } else {
+    await boss.unschedule(JOB_INVOICE_VERIFY).catch(() => {});
+  }
+
+  await boss.schedule(JOB_INTEGRATION_OUTBOX, fastify.config.integrationOutboxCron, null, {
+    tz: fastify.config.appTimezone,
+    singletonKey: JOB_INTEGRATION_OUTBOX,
+    singletonSeconds: 55,
+  });
+
  fastify.log.info(
    {
      schema: fastify.config.pgBossSchema,
      ocrProvider: fastify.config.ocrProvider,
      ocrCron: fastify.config.ocrPollCron,
+      verificationEnabled: fastify.config.invoiceVerificationEnabled,
+      verificationCron: fastify.config.verificationPollCron,
      delegationCron: fastify.config.delegationRevokeCron,
      invoiceArchiveCron: fastify.config.invoiceArchiveCron,
+      integrationOutboxCron: fastify.config.integrationOutboxCron,
    },
    "pg-boss job system started",
  );
@@ -189,5 +304,7 @@ module.exports = {
  JOB_OCR_TICK,
  JOB_DELEGATION_REVOKE,
  JOB_INVOICE_ARCHIVE,
+  JOB_INVOICE_VERIFY,
+  JOB_INTEGRATION_OUTBOX,
  startPgBossJobs,
 };
@@ -0,0 +1,96 @@
+const { verifyInvoiceByProvider } = require("../services/verification");
+
+function toResultStatus(status) {
+  if (["passed", "failed", "error"].includes(status)) {
+    return status;
+  }
+
+  return "error";
+}
+
+async function runVerificationForRecord(prisma, config, recordId) {
+  const result = await prisma.$transaction(async (tx) => {
+    const lockedRows = await tx.$queryRaw`
+      SELECT id
+      FROM "invoice_verification_records"
+      WHERE id = ${recordId}::uuid
+      FOR UPDATE
+    `;
+
+    if (!Array.isArray(lockedRows) || lockedRows.length === 0) {
+      return { skipped: true, reason: "record_not_found" };
+    }
+
+    const record = await tx.invoiceVerificationRecord.findUnique({
+      where: { id: recordId },
+      include: {
+        invoice: {
+          select: {
+            id: true,
+            invoice_code: true,
+            invoice_number: true,
+            total_amount: true,
+            remarks: true,
+            status: true,
+          },
+        },
+      },
+    });
+
+    if (!record || !record.invoice) {
+      await tx.invoiceVerificationRecord.update({
+        where: { id: recordId },
+        data: {
+          status: "error",
+          error_code: "invoice_not_found",
+          error_message: "invoice not found",
+          completed_at: new Date(),
+        },
+      });
+      return { skipped: true, reason: "invoice_not_found" };
+    }
+
+    if (record.status !== "pending") {
+      return { skipped: true, reason: "status_not_pending" };
+    }
+
+    const startedAt = Date.now();
+    const verification = await verifyInvoiceByProvider({
+      config,
+      invoice: record.invoice,
+    });
+    const durationMs = Date.now() - startedAt;
+
+    const status = toResultStatus(verification.status);
+
+    await tx.invoiceVerificationRecord.update({
+      where: { id: recordId },
+      data: {
+        status,
+        result_code: verification.resultCode ?? null,
+        result_message: verification.resultMessage ?? null,
+        result_payload: verification.payload ?? null,
+        error_code: status === "error" ? verification.resultCode ?? "verification_error" : null,
+        error_message:
+          status === "error"
+            ? verification.resultMessage ?? "verification_error"
+            : null,
+        duration_ms: durationMs,
+        completed_at: new Date(),
+      },
+    });
+
+    return {
+      skipped: false,
+      status,
+      durationMs,
+      resultCode: verification.resultCode ?? null,
+    };
+  });
+
+  return result;
+}
+
+module.exports = {
+  runVerificationForRecord,
+};