hyb-oa/server/src/jobs/pgBossJobs.js

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 {
  FEATURE_FLAG_KEYS,
  evaluateFeatureFlag,
} = require("../services/featureFlags");

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;
  if (!connectionString) {
    throw new Error("Missing DATABASE_URL");
  }
  return connectionString;
}

async function startPgBossJobs(fastify) {
  const connectionString = resolveDatabaseConnectionString();

  const boss = new PgBoss({
    connectionString,
    schema: fastify.config.pgBossSchema,
  });

  await boss.start();

  fastify.decorate("boss", boss);

  fastify.addHook("onClose", async () => {
    await boss.stop({ graceful: true, timeout: 10_000 });
  });

  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];
    try {
      const revoked = await revokeExpiredDelegations(fastify.prisma, fastify.log);
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_DELEGATION_REVOKE,
        driver: "pgboss",
        outcome: "success",
      });
      if (revoked > 0) {
        fastify.metrics?.jobProcessedTotal?.inc?.(
          { job: JOB_DELEGATION_REVOKE, driver: "pgboss", kind: "revoked" },
          revoked,
        );
        fastify.log.info(
          { jobId: job?.id, revoked },
          "pg-boss delegation revoke tick processed",
        );
      }
    } catch (error) {
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_DELEGATION_REVOKE,
        driver: "pgboss",
        outcome: "failure",
      });
      fastify.log.error({ err: error, jobId: job?.id }, "pg-boss delegation revoke tick failed");
      throw error;
    }
  });

  await boss.work(JOB_INVOICE_ARCHIVE, { batchSize: 1 }, async (jobs) => {
    const job = jobs?.[0];
    try {
      const archived = await archiveUnusedInvoices(fastify.prisma, fastify.log, {
        batchSize: fastify.config.invoiceArchiveBatchSize,
        unusedDays: fastify.config.invoiceArchiveUnusedDays,
      });
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_INVOICE_ARCHIVE,
        driver: "pgboss",
        outcome: "success",
      });
      if (archived > 0) {
        fastify.metrics?.jobProcessedTotal?.inc?.(
          { job: JOB_INVOICE_ARCHIVE, driver: "pgboss", kind: "archived" },
          archived,
        );
        fastify.log.info(
          { jobId: job?.id, archived, unusedDays: fastify.config.invoiceArchiveUnusedDays },
          "pg-boss invoice archive tick processed",
        );
      }
    } catch (error) {
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_INVOICE_ARCHIVE,
        driver: "pgboss",
        outcome: "failure",
      });
      fastify.log.error({ err: error, jobId: job?.id }, "pg-boss invoice archive tick failed");
      throw error;
    }
  });

  await boss.work(JOB_OCR_TICK, { batchSize: 1 }, async (jobs) => {
    if (!["tencent", "mock"].includes(fastify.config.ocrProvider)) {
      return;
    }

    const job = jobs?.[0];
    const workerId = `pgboss:ocr:${process.pid}`;
    try {
      const stats = await runOcrOnce(fastify.prisma, fastify.config, fastify.log, workerId);
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_OCR_TICK,
        driver: "pgboss",
        outcome: "success",
      });
      if (stats.available > 0) {
        fastify.metrics?.jobProcessedTotal?.inc?.(
          { job: JOB_OCR_TICK, driver: "pgboss", kind: "available" },
          stats.available,
        );
      }
      if (stats.duplicate > 0) {
        fastify.metrics?.jobProcessedTotal?.inc?.(
          { job: JOB_OCR_TICK, driver: "pgboss", kind: "duplicate" },
          stats.duplicate,
        );
      }
      if (stats.failed > 0) {
        fastify.metrics?.jobProcessedTotal?.inc?.(
          { job: JOB_OCR_TICK, driver: "pgboss", kind: "failed" },
          stats.failed,
        );
      }

      if (stats.claimed > 0) {
        fastify.log.info(
          { jobId: job?.id, workerId, ...stats },
          "pg-boss OCR tick processed",
        );
      }
    } catch (error) {
      fastify.metrics?.jobRunsTotal?.inc?.({
        job: JOB_OCR_TICK,
        driver: "pgboss",
        outcome: "failure",
      });
      fastify.log.error({ err: error, jobId: job?.id }, "pg-boss OCR tick failed");
      throw error;
    }
  });

  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,
    singletonMinutes: 60,
  });

  await boss.schedule(JOB_INVOICE_ARCHIVE, fastify.config.invoiceArchiveCron, null, {
    tz: fastify.config.appTimezone,
    singletonKey: JOB_INVOICE_ARCHIVE,
    singletonHours: 24,
  });

  if (["tencent", "mock"].includes(fastify.config.ocrProvider)) {
    await boss.schedule(JOB_OCR_TICK, fastify.config.ocrPollCron, null, {
      tz: fastify.config.appTimezone,
      singletonKey: JOB_OCR_TICK,
      singletonSeconds: 55,
    });
  } else {
    await boss.unschedule(JOB_OCR_TICK).catch(() => {});
  }

  const verificationScheduling = await evaluateFeatureFlag({
    prisma: fastify.prisma,
    config: fastify.config,
    flagKey: FEATURE_FLAG_KEYS.invoiceVerification,
    strictDepartment: false,
  });

  if (verificationScheduling.enabled) {
    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: verificationScheduling.enabled,
      verificationFlagReason: verificationScheduling.reason,
      verificationCron: fastify.config.verificationPollCron,
      delegationCron: fastify.config.delegationRevokeCron,
      invoiceArchiveCron: fastify.config.invoiceArchiveCron,
      integrationOutboxCron: fastify.config.integrationOutboxCron,
    },
    "pg-boss job system started",
  );

  return boss;
}

module.exports = {
  JOB_OCR_TICK,
  JOB_DELEGATION_REVOKE,
  JOB_INVOICE_ARCHIVE,
  JOB_INVOICE_VERIFY,
  JOB_INTEGRATION_OUTBOX,
  startPgBossJobs,
};