feat: 完成P10生产硬化与矩阵收口（P10E1-P10G1）

server/src/jobs/integrationOutboxJob.js

@@ -0,0 +1,124 @@
+const cron = require("node-cron");
+
+const {
+  fetchPendingOutboxEvents,
+  markOutboxFailed,
+  markOutboxProcessing,
+  markOutboxSucceeded,
+} = require("../services/integrations/outbox");
+const { dispatchOutboxEvent } = require("../services/integrations/dispatcher");
+
+async function runIntegrationOutboxOnce(prisma, config, logger) {
+  const items = await fetchPendingOutboxEvents(prisma, {
+    batchSize: config.integrationOutboxBatchSize,
+  });
+
+  let succeeded = 0;
+  let failed = 0;
+
+  for (const item of items) {
+    await markOutboxProcessing(prisma, item.id, null);
+
+    try {
+      await dispatchOutboxEvent({
+        config,
+        eventType: item.event_type,
+        payload: item.payload,
+      });
+      await markOutboxSucceeded(prisma, item.id, null);
+      succeeded += 1;
+    } catch (error) {
+      const retryable = typeof error?.retryable === "boolean" ? error.retryable : true;
+      await markOutboxFailed(prisma, config, item, error, null, { retryable });
+      failed += 1;
+      logger.error(
+        {
+          err: error,
+          retryable,
+          outboxId: item.id,
+          eventType: item.event_type,
+        },
+        "Integration outbox dispatch failed",
+      );
+    }
+  }
+
+  return {
+    total: items.length,
+    succeeded,
+    failed,
+  };
+}
+
+function startIntegrationOutboxJob(fastify) {
+  let running = false;
+
+  const task = cron.schedule(fastify.config.integrationOutboxCron, async () => {
+    if (running) {
+      return;
+    }
+
+    running = true;
+    try {
+      const result = await runIntegrationOutboxOnce(
+        fastify.prisma,
+        fastify.config,
+        fastify.log,
+      );
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: "integration.outbox.dispatch",
+        driver: "cron",
+        outcome: "success",
+      });
+      if (result.succeeded > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          {
+            job: "integration.outbox.dispatch",
+            driver: "cron",
+            kind: "succeeded",
+          },
+          result.succeeded,
+        );
+      }
+      if (result.failed > 0) {
+        fastify.metrics?.jobProcessedTotal?.inc?.(
+          {
+            job: "integration.outbox.dispatch",
+            driver: "cron",
+            kind: "failed",
+          },
+          result.failed,
+        );
+      }
+    } catch (error) {
+      fastify.log.error({ err: error }, "Integration outbox job failed");
+      fastify.metrics?.jobRunsTotal?.inc?.({
+        job: "integration.outbox.dispatch",
+        driver: "cron",
+        outcome: "failure",
+      });
+    } finally {
+      running = false;
+    }
+  });
+
+  fastify.addHook("onClose", async () => {
+    task.stop();
+    task.destroy();
+  });
+
+  fastify.log.info(
+    {
+      cron: fastify.config.integrationOutboxCron,
+      batchSize: fastify.config.integrationOutboxBatchSize,
+    },
+    "Integration outbox job scheduled",
+  );
+
+  return task;
+}
+
+module.exports = {
+  runIntegrationOutboxOnce,
+  startIntegrationOutboxJob,
+};