forked from wangziqi/gongxue-base
feat: migrate backend foundation to NestJS
This commit is contained in:
@@ -25,6 +25,8 @@
|
||||
"student-supervision:once": "tsx src/index.ts --once --job student-supervision"
|
||||
},
|
||||
"dependencies": {
|
||||
"@nestjs/common": "^11.1.28",
|
||||
"@nestjs/core": "^11.1.28",
|
||||
"@resvg/resvg-js": "^2.6.2",
|
||||
"@supabase/storage-js": "^2.108.2",
|
||||
"ali-oss": "^6.23.0",
|
||||
@@ -32,7 +34,9 @@
|
||||
"iconv-lite": "^0.6.3",
|
||||
"jszip": "^3.10.1",
|
||||
"pdfkit": "^0.19.1",
|
||||
"pg": "^8.16.3"
|
||||
"pg": "^8.16.3",
|
||||
"reflect-metadata": "^0.2.2",
|
||||
"rxjs": "^7.8.2"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^24.0.4",
|
||||
|
||||
@@ -1,210 +1,13 @@
|
||||
import { closePool } from './db.js';
|
||||
import { config } from './config.js';
|
||||
import { processCrmBatch } from './jobs/crm.js';
|
||||
import { processCommerceBatch } from './jobs/commerce.js';
|
||||
import { processAssetBatch } from './jobs/assets.js';
|
||||
import {
|
||||
parseWorkerCli,
|
||||
type ContinuousWorkerJob,
|
||||
type WorkerJob,
|
||||
} from './cli.js';
|
||||
|
||||
const extraClosers = new Set<() => Promise<void>>();
|
||||
|
||||
async function runOnce(job: WorkerJob, month?: string) {
|
||||
if (job === 'crm') {
|
||||
const result = await processCrmBatch();
|
||||
console.log(`[worker] crm batch processed=${result.processed} sent=${result.sent} failed=${result.failed} retrying=${result.retrying} discarded=${result.discarded}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'commerce') {
|
||||
const result = await processCommerceBatch();
|
||||
console.log(
|
||||
`[worker] commerce batch processed=${result.processed}`
|
||||
+ ` payments=${result.payments.processed} paid=${result.payments.paid} pending=${result.payments.pending} closed=${result.payments.closed} failed=${result.payments.failed} paymentErrors=${result.payments.errors}`
|
||||
+ ` refunds=${result.refunds.processed} succeeded=${result.refunds.succeeded} processing=${result.refunds.processing} refundFailed=${result.refunds.failed} refundErrors=${result.refunds.errors}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'provider-bills') {
|
||||
const { closePool: closeApiPool } = await import('../../api/src/core/db.js');
|
||||
const { processProviderBillBatch } = await import('./jobs/provider-bills.js');
|
||||
extraClosers.add(closeApiPool);
|
||||
const result = await processProviderBillBatch();
|
||||
console.log(
|
||||
`[worker] provider-bills batch processed=${result.processed}`
|
||||
+ ` completed=${result.completed} failed=${result.failed} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-billing') {
|
||||
const { processPlatformBillingBatch } = await import('./jobs/platform-billing.js');
|
||||
const result = await processPlatformBillingBatch();
|
||||
console.log(
|
||||
`[worker] platform-billing batch processed=${result.processed}`
|
||||
+ ` created=${result.created} skipped=${result.skipped} failed=${result.failed}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-usage') {
|
||||
const { processPlatformUsageBatch } = await import('./jobs/platform-usage.js');
|
||||
const result = await processPlatformUsageBatch({ month });
|
||||
console.log(
|
||||
`[worker] platform-usage batch processed=${result.processed}`
|
||||
+ ` metrics=${result.metrics} created=${result.created}`
|
||||
+ ` updated=${result.updated} failed=${result.failed} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-usage-overage') {
|
||||
const { processPlatformUsageOverageBatch } = await import('./jobs/platform-usage-overage.js');
|
||||
const result = await processPlatformUsageOverageBatch({ month });
|
||||
console.log(
|
||||
`[worker] platform-usage-overage batch processed=${result.processed}`
|
||||
+ ` created=${result.created} skipped=${result.skipped}`
|
||||
+ ` failed=${result.failed} totalCents=${result.totalCents}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-dunning') {
|
||||
const { processPlatformDunningBatch } = await import('./jobs/platform-dunning.js');
|
||||
const result = await processPlatformDunningBatch();
|
||||
console.log(
|
||||
`[worker] platform-dunning batch processed=${result.processed}`
|
||||
+ ` markedOverdue=${result.markedOverdue} reminderCreated=${result.reminderCreated}`
|
||||
+ ` skippedReminder=${result.skippedReminder}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-dunning-notifications') {
|
||||
const { processPlatformDunningNotificationBatch } = await import('./jobs/platform-dunning-notifications.js');
|
||||
const result = await processPlatformDunningNotificationBatch();
|
||||
console.log(
|
||||
`[worker] platform-dunning-notifications batch enqueued=${result.enqueued}`
|
||||
+ ` processed=${result.processed} sent=${result.sent} failed=${result.failed}`
|
||||
+ ` retrying=${result.retrying} discarded=${result.discarded}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-audit-alerts') {
|
||||
const { processPlatformAuditAlertBatch } = await import('./jobs/platform-audit-alerts.js');
|
||||
const result = await processPlatformAuditAlertBatch();
|
||||
console.log(
|
||||
`[worker] platform-audit-alerts batch processed=${result.processed}`
|
||||
+ ` created=${result.created} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-audit-notifications') {
|
||||
const { processPlatformAuditNotificationBatch } = await import('./jobs/platform-audit-notifications.js');
|
||||
const result = await processPlatformAuditNotificationBatch();
|
||||
console.log(
|
||||
`[worker] platform-audit-notifications batch enqueued=${result.enqueued}`
|
||||
+ ` processed=${result.processed} sent=${result.sent} failed=${result.failed}`
|
||||
+ ` retrying=${result.retrying} discarded=${result.discarded}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'assets') {
|
||||
const result = await processAssetBatch();
|
||||
console.log(
|
||||
`[worker] assets batch processed=${result.processed}`
|
||||
+ ` verified=${result.verified} failed=${result.failed} skipped=${result.skipped} errors=${result.errors}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'imports') {
|
||||
const { closeImportExecutorPool, processImportBatch } = await import('./jobs/imports.js');
|
||||
extraClosers.add(closeImportExecutorPool);
|
||||
const result = await processImportBatch();
|
||||
console.log(
|
||||
`[worker] imports batch processed=${result.processed}`
|
||||
+ ` completed=${result.completed} completedWithErrors=${result.completedWithErrors}`
|
||||
+ ` failed=${result.failed} retrying=${result.retrying}`
|
||||
+ ` leaseLost=${result.leaseLost} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'public-banks') {
|
||||
const { closePublicBankSyncExecutorPool, processPublicBankSyncBatch } = await import('./jobs/public-banks.js');
|
||||
extraClosers.add(closePublicBankSyncExecutorPool);
|
||||
const result = await processPublicBankSyncBatch();
|
||||
console.log(
|
||||
`[worker] public-banks batch processed=${result.processed}`
|
||||
+ ` synced=${result.synced} conflicts=${result.conflicts}`
|
||||
+ ` failed=${result.failed} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'exports') {
|
||||
const { closeExportExecutorPool, processExportBatch } = await import('./jobs/exports.js');
|
||||
extraClosers.add(closeExportExecutorPool);
|
||||
const result = await processExportBatch();
|
||||
console.log(
|
||||
`[worker] exports batch processed=${result.processed}`
|
||||
+ ` completed=${result.completed} failed=${result.failed}`
|
||||
+ ` retrying=${result.retrying} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'student-supervision') {
|
||||
const { closePool: closeApiPool } = await import('../../api/src/core/db.js');
|
||||
const { processStudentSupervisionBatch } = await import('./jobs/student-supervision.js');
|
||||
extraClosers.add(closeApiPool);
|
||||
const result = await processStudentSupervisionBatch();
|
||||
console.log(
|
||||
`[worker] student-supervision batch processed=${result.processed}`
|
||||
+ ` generated=${result.generated} followups=${result.followups}`
|
||||
+ ` failed=${result.failed} skipped=${result.skipped}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
throw new Error(`Unsupported worker job: ${job}`);
|
||||
}
|
||||
|
||||
function loopPollIntervalMs(job: ContinuousWorkerJob) {
|
||||
if (job === 'crm') return config.crmPollIntervalMs;
|
||||
if (job === 'commerce') return config.commercePollIntervalMs;
|
||||
if (job === 'provider-bills') return config.providerBillPollIntervalMs;
|
||||
if (job === 'platform-dunning-notifications') return config.platformDunningNotificationPollIntervalMs;
|
||||
if (job === 'platform-audit-notifications') return config.platformAuditNotificationPollIntervalMs;
|
||||
if (job === 'assets') return config.assetPollIntervalMs;
|
||||
if (job === 'imports') return config.importPollIntervalMs;
|
||||
if (job === 'public-banks') return config.publicBankSyncPollIntervalMs;
|
||||
return config.exportPollIntervalMs;
|
||||
}
|
||||
|
||||
async function runLoop(job: ContinuousWorkerJob) {
|
||||
const pollIntervalMs = loopPollIntervalMs(job);
|
||||
console.log(`[worker] started job=${job} pollIntervalMs=${pollIntervalMs}`);
|
||||
let stopped = false;
|
||||
const stop = () => {
|
||||
stopped = true;
|
||||
};
|
||||
process.once('SIGINT', stop);
|
||||
process.once('SIGTERM', stop);
|
||||
|
||||
while (!stopped) {
|
||||
try {
|
||||
await runOnce(job);
|
||||
} catch (error) {
|
||||
console.error(`[worker] job=${job} failed`, error);
|
||||
}
|
||||
if (!stopped) await new Promise(resolve => setTimeout(resolve, pollIntervalMs));
|
||||
}
|
||||
}
|
||||
import 'reflect-metadata';
|
||||
import { NestFactory } from '@nestjs/core';
|
||||
import { parseWorkerCli } from './cli.js';
|
||||
import { WorkerModule, WorkerRunner } from './worker.module.js';
|
||||
|
||||
const cli = parseWorkerCli(process.argv.slice(2));
|
||||
const app = await NestFactory.createApplicationContext(WorkerModule, { logger: false });
|
||||
|
||||
try {
|
||||
if (cli.loop) {
|
||||
await runLoop(cli.job as ContinuousWorkerJob);
|
||||
} else {
|
||||
await runOnce(cli.job, cli.month);
|
||||
}
|
||||
await app.get(WorkerRunner).execute(cli);
|
||||
} finally {
|
||||
for (const closeExtra of extraClosers) {
|
||||
await closeExtra();
|
||||
}
|
||||
await closePool();
|
||||
await app.close();
|
||||
}
|
||||
|
||||
161
apps/worker/src/worker.module.ts
Normal file
161
apps/worker/src/worker.module.ts
Normal file
@@ -0,0 +1,161 @@
|
||||
import { Injectable, Module, OnApplicationShutdown } from '@nestjs/common';
|
||||
import { config } from './config.js';
|
||||
import { closePool } from './db.js';
|
||||
import { type ContinuousWorkerJob, type WorkerCliOptions, type WorkerJob } from './cli.js';
|
||||
import { processAssetBatch } from './jobs/assets.js';
|
||||
import { processCommerceBatch } from './jobs/commerce.js';
|
||||
import { processCrmBatch } from './jobs/crm.js';
|
||||
|
||||
@Injectable()
|
||||
export class WorkerCloserRegistry implements OnApplicationShutdown {
|
||||
// 动态任务可能额外加载 API 或导入导出连接池，统一登记后由 Nest 生命周期关闭。
|
||||
private readonly closers = new Set<() => Promise<void>>([closePool]);
|
||||
|
||||
add(closer: () => Promise<void>) { this.closers.add(closer); }
|
||||
|
||||
async onApplicationShutdown() {
|
||||
for (const closer of this.closers) await closer();
|
||||
}
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class WorkerJobRegistry {
|
||||
constructor(private readonly closers: WorkerCloserRegistry) {}
|
||||
|
||||
async run(job: WorkerJob, month?: string) {
|
||||
if (job === 'crm') {
|
||||
const result = await processCrmBatch();
|
||||
console.log(`[worker] crm batch processed=${result.processed} sent=${result.sent} failed=${result.failed} retrying=${result.retrying} discarded=${result.discarded}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'commerce') {
|
||||
const result = await processCommerceBatch();
|
||||
console.log(`[worker] commerce batch processed=${result.processed} payments=${result.payments.processed} paid=${result.payments.paid} pending=${result.payments.pending} closed=${result.payments.closed} failed=${result.payments.failed} paymentErrors=${result.payments.errors} refunds=${result.refunds.processed} succeeded=${result.refunds.succeeded} processing=${result.refunds.processing} refundFailed=${result.refunds.failed} refundErrors=${result.refunds.errors}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'provider-bills') {
|
||||
const { closePool: closeApiPool } = await import('../../api/src/core/db.js');
|
||||
const { processProviderBillBatch } = await import('./jobs/provider-bills.js');
|
||||
this.closers.add(closeApiPool);
|
||||
const result = await processProviderBillBatch();
|
||||
console.log(`[worker] provider-bills batch processed=${result.processed} completed=${result.completed} failed=${result.failed} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-billing') {
|
||||
const { processPlatformBillingBatch } = await import('./jobs/platform-billing.js');
|
||||
const result = await processPlatformBillingBatch();
|
||||
console.log(`[worker] platform-billing batch processed=${result.processed} created=${result.created} skipped=${result.skipped} failed=${result.failed}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-usage') {
|
||||
const { processPlatformUsageBatch } = await import('./jobs/platform-usage.js');
|
||||
const result = await processPlatformUsageBatch({ month });
|
||||
console.log(`[worker] platform-usage batch processed=${result.processed} metrics=${result.metrics} created=${result.created} updated=${result.updated} failed=${result.failed} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-usage-overage') {
|
||||
const { processPlatformUsageOverageBatch } = await import('./jobs/platform-usage-overage.js');
|
||||
const result = await processPlatformUsageOverageBatch({ month });
|
||||
console.log(`[worker] platform-usage-overage batch processed=${result.processed} created=${result.created} skipped=${result.skipped} failed=${result.failed} totalCents=${result.totalCents}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-dunning') {
|
||||
const { processPlatformDunningBatch } = await import('./jobs/platform-dunning.js');
|
||||
const result = await processPlatformDunningBatch();
|
||||
console.log(`[worker] platform-dunning batch processed=${result.processed} markedOverdue=${result.markedOverdue} reminderCreated=${result.reminderCreated} skippedReminder=${result.skippedReminder}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-dunning-notifications') {
|
||||
const { processPlatformDunningNotificationBatch } = await import('./jobs/platform-dunning-notifications.js');
|
||||
const result = await processPlatformDunningNotificationBatch();
|
||||
console.log(`[worker] platform-dunning-notifications batch enqueued=${result.enqueued} processed=${result.processed} sent=${result.sent} failed=${result.failed} retrying=${result.retrying} discarded=${result.discarded}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-audit-alerts') {
|
||||
const { processPlatformAuditAlertBatch } = await import('./jobs/platform-audit-alerts.js');
|
||||
const result = await processPlatformAuditAlertBatch();
|
||||
console.log(`[worker] platform-audit-alerts batch processed=${result.processed} created=${result.created} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-audit-notifications') {
|
||||
const { processPlatformAuditNotificationBatch } = await import('./jobs/platform-audit-notifications.js');
|
||||
const result = await processPlatformAuditNotificationBatch();
|
||||
console.log(`[worker] platform-audit-notifications batch enqueued=${result.enqueued} processed=${result.processed} sent=${result.sent} failed=${result.failed} retrying=${result.retrying} discarded=${result.discarded}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'assets') {
|
||||
const result = await processAssetBatch();
|
||||
console.log(`[worker] assets batch processed=${result.processed} verified=${result.verified} failed=${result.failed} skipped=${result.skipped} errors=${result.errors}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'imports') {
|
||||
const { closeImportExecutorPool, processImportBatch } = await import('./jobs/imports.js');
|
||||
this.closers.add(closeImportExecutorPool);
|
||||
const result = await processImportBatch();
|
||||
console.log(`[worker] imports batch processed=${result.processed} completed=${result.completed} completedWithErrors=${result.completedWithErrors} failed=${result.failed} retrying=${result.retrying} leaseLost=${result.leaseLost} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'public-banks') {
|
||||
const { closePublicBankSyncExecutorPool, processPublicBankSyncBatch } = await import('./jobs/public-banks.js');
|
||||
this.closers.add(closePublicBankSyncExecutorPool);
|
||||
const result = await processPublicBankSyncBatch();
|
||||
console.log(`[worker] public-banks batch processed=${result.processed} synced=${result.synced} conflicts=${result.conflicts} failed=${result.failed} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'exports') {
|
||||
const { closeExportExecutorPool, processExportBatch } = await import('./jobs/exports.js');
|
||||
this.closers.add(closeExportExecutorPool);
|
||||
const result = await processExportBatch();
|
||||
console.log(`[worker] exports batch processed=${result.processed} completed=${result.completed} failed=${result.failed} retrying=${result.retrying} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
if (job === 'student-supervision') {
|
||||
const { closePool: closeApiPool } = await import('../../api/src/core/db.js');
|
||||
const { processStudentSupervisionBatch } = await import('./jobs/student-supervision.js');
|
||||
this.closers.add(closeApiPool);
|
||||
const result = await processStudentSupervisionBatch();
|
||||
console.log(`[worker] student-supervision batch processed=${result.processed} generated=${result.generated} followups=${result.followups} failed=${result.failed} skipped=${result.skipped}`);
|
||||
return;
|
||||
}
|
||||
throw new Error(`Unsupported worker job: ${job}`);
|
||||
}
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class WorkerPollIntervals {
|
||||
get(job: ContinuousWorkerJob) {
|
||||
if (job === 'crm') return config.crmPollIntervalMs;
|
||||
if (job === 'commerce') return config.commercePollIntervalMs;
|
||||
if (job === 'provider-bills') return config.providerBillPollIntervalMs;
|
||||
if (job === 'platform-dunning-notifications') return config.platformDunningNotificationPollIntervalMs;
|
||||
if (job === 'platform-audit-notifications') return config.platformAuditNotificationPollIntervalMs;
|
||||
if (job === 'assets') return config.assetPollIntervalMs;
|
||||
if (job === 'imports') return config.importPollIntervalMs;
|
||||
if (job === 'public-banks') return config.publicBankSyncPollIntervalMs;
|
||||
return config.exportPollIntervalMs;
|
||||
}
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class WorkerRunner {
|
||||
constructor(private readonly jobs: WorkerJobRegistry, private readonly intervals: WorkerPollIntervals) {}
|
||||
|
||||
async execute(cli: WorkerCliOptions) {
|
||||
if (!cli.loop) return this.jobs.run(cli.job, cli.month);
|
||||
const job = cli.job as ContinuousWorkerJob;
|
||||
const pollIntervalMs = this.intervals.get(job);
|
||||
console.log(`[worker] started job=${job} pollIntervalMs=${pollIntervalMs}`);
|
||||
let stopped = false;
|
||||
const stop = () => { stopped = true; };
|
||||
process.once('SIGINT', stop);
|
||||
process.once('SIGTERM', stop);
|
||||
while (!stopped) {
|
||||
try { await this.jobs.run(job); }
|
||||
catch (error) { console.error(`[worker] job=${job} failed`, error); }
|
||||
if (!stopped) await new Promise(resolve => setTimeout(resolve, pollIntervalMs));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Module({ providers: [WorkerCloserRegistry, WorkerJobRegistry, WorkerPollIntervals, WorkerRunner] })
|
||||
export class WorkerModule {}
|
||||
@@ -4,6 +4,8 @@
|
||||
"module": "NodeNext",
|
||||
"moduleResolution": "NodeNext",
|
||||
"strict": true,
|
||||
"experimentalDecorators": true,
|
||||
"emitDecoratorMetadata": true,
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
"outDir": "dist",
|
||||
|
||||
Reference in New Issue
Block a user