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 JOB_OCR_TICK = "ocr.tick";
const JOB_DELEGATION_REVOKE = "delegation.revoke-expired";
const JOB_INVOICE_ARCHIVE = "invoice.archive-unused";

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.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.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(() => {});
  }

  fastify.log.info(
    {
      schema: fastify.config.pgBossSchema,
      ocrProvider: fastify.config.ocrProvider,
      ocrCron: fastify.config.ocrPollCron,
      delegationCron: fastify.config.delegationRevokeCron,
      invoiceArchiveCron: fastify.config.invoiceArchiveCron,
    },
    "pg-boss job system started",
  );

  return boss;
}

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