feat(server): 完成 F19~F19d pg-boss 作业系统迁移\n\n- 引入 pg-boss(JOB_DRIVER=pgboss)并保留 cron 回滚\n- OCR/委派回收/超期归档改由 pg-boss schedule + work 执行\n- OCR_PROVIDER=off 自动 unschedule ocr.tick\n- 更新 trialrun env/compose/runbook 与 README,并推进 CSV 状态为 DONE
这个提交包含在:
+2
-6
@@ -29,9 +29,7 @@ const configReadRoutes = require("./routes/configRead");
|
||||
const approvalRulesRoutes = require("./routes/approvalRules");
|
||||
const notificationsRoutes = require("./routes/notifications");
|
||||
const exportsRoutes = require("./routes/exports");
|
||||
const { startDelegationRevokeJob } = require("./jobs/delegationRevokeJob");
|
||||
const { startOcrJob } = require("./jobs/ocrJob");
|
||||
const { startInvoiceArchiveJob } = require("./jobs/invoiceArchiveJob");
|
||||
const { startJobs } = require("./jobs/jobSystem");
|
||||
const { AppError } = require("./utils/AppError");
|
||||
|
||||
async function buildApp() {
|
||||
@@ -207,9 +205,7 @@ async function buildApp() {
|
||||
await app.register(notificationsRoutes, { prefix: "/api" });
|
||||
await app.register(exportsRoutes, { prefix: "/api" });
|
||||
|
||||
startDelegationRevokeJob(app);
|
||||
startOcrJob(app);
|
||||
startInvoiceArchiveJob(app);
|
||||
await startJobs(app);
|
||||
|
||||
return app;
|
||||
}
|
||||
|
||||
@@ -62,6 +62,10 @@ function getConfig() {
|
||||
|
||||
delegationRevokeCron: process.env.DELEGATION_REVOKE_CRON ?? "0 * * * *",
|
||||
|
||||
jobDriver: process.env.JOB_DRIVER ?? "cron",
|
||||
jobsEnabled: process.env.JOBS_ENABLED !== "false",
|
||||
pgBossSchema: process.env.PG_BOSS_SCHEMA ?? "pgboss",
|
||||
|
||||
ocrProvider: process.env.OCR_PROVIDER ?? "off",
|
||||
ocrBatchSize: Number(process.env.OCR_BATCH_SIZE ?? 10),
|
||||
ocrLockTtlSeconds: Number(process.env.OCR_LOCK_TTL_SECONDS ?? 300),
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
const { startDelegationRevokeJob } = require("./delegationRevokeJob");
|
||||
const { startInvoiceArchiveJob } = require("./invoiceArchiveJob");
|
||||
const { startOcrJob } = require("./ocrJob");
|
||||
const { startPgBossJobs } = require("./pgBossJobs");
|
||||
|
||||
async function startJobs(fastify) {
|
||||
if (!fastify.config.jobsEnabled) {
|
||||
fastify.log.info("Jobs disabled");
|
||||
return { driver: "disabled" };
|
||||
}
|
||||
|
||||
if (fastify.config.jobDriver === "pgboss") {
|
||||
await startPgBossJobs(fastify);
|
||||
return { driver: "pgboss" };
|
||||
}
|
||||
|
||||
startDelegationRevokeJob(fastify);
|
||||
startOcrJob(fastify);
|
||||
startInvoiceArchiveJob(fastify);
|
||||
return { driver: "cron" };
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
startJobs,
|
||||
};
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
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];
|
||||
const revoked = await revokeExpiredDelegations(fastify.prisma, fastify.log);
|
||||
if (revoked > 0) {
|
||||
fastify.log.info(
|
||||
{ jobId: job?.id, revoked },
|
||||
"pg-boss delegation revoke tick processed",
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
await boss.work(JOB_INVOICE_ARCHIVE, { batchSize: 1 }, async (jobs) => {
|
||||
const job = jobs?.[0];
|
||||
const archived = await archiveUnusedInvoices(fastify.prisma, fastify.log, {
|
||||
batchSize: fastify.config.invoiceArchiveBatchSize,
|
||||
unusedDays: fastify.config.invoiceArchiveUnusedDays,
|
||||
});
|
||||
if (archived > 0) {
|
||||
fastify.log.info(
|
||||
{ jobId: job?.id, archived, unusedDays: fastify.config.invoiceArchiveUnusedDays },
|
||||
"pg-boss invoice archive tick processed",
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
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}`;
|
||||
const stats = await runOcrOnce(fastify.prisma, fastify.config, fastify.log, workerId);
|
||||
if (stats.claimed > 0) {
|
||||
fastify.log.info(
|
||||
{ jobId: job?.id, workerId, ...stats },
|
||||
"pg-boss OCR tick processed",
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
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,
|
||||
};
|
||||
在新工单中引用
屏蔽一个用户