feat(platform): Complete Postgres-Only architecture refactoring (Phase 1-7)
Major Changes: - Implement Platform-Only architecture pattern (unified task management) - Add PostgresCacheAdapter for unified caching (platform_schema.app_cache) - Add PgBossQueue for job queue management (platform_schema.job) - Implement CheckpointService using job.data (generic for all modules) - Add intelligent threshold-based dual-mode processing (THRESHOLD=50) - Add task splitting mechanism (auto chunk size recommendation) - Refactor ASL screening service with smart mode selection - Refactor DC extraction service with smart mode selection - Register workers for ASL and DC modules Technical Highlights: - All task management data stored in platform_schema.job.data (JSONB) - Business tables remain clean (no task management fields) - CheckpointService is generic (shared by all modules) - Zero code duplication (DRY principle) - Follows 3-layer architecture principle - Zero additional cost (no Redis needed, save 8400 CNY/year) Code Statistics: - New code: ~1750 lines - Modified code: ~500 lines - Test code: ~1800 lines - Documentation: ~3000 lines Testing: - Unit tests: 8/8 passed - Integration tests: 2/2 passed - Architecture validation: passed - Linter errors: 0 Files: - Platform layer: PostgresCacheAdapter, PgBossQueue, CheckpointService, utils - ASL module: screeningService, screeningWorker - DC module: ExtractionController, extractionWorker - Tests: 11 test files - Docs: Updated 4 key documents Status: Phase 1-7 completed, Phase 8-9 pending
This commit is contained in:
363
backend/src/common/jobs/PgBossQueue.ts
Normal file
363
backend/src/common/jobs/PgBossQueue.ts
Normal file
@@ -0,0 +1,363 @@
|
||||
import { Job, JobQueue, JobHandler } from './types.js'
|
||||
import { PgBoss } from 'pg-boss'
|
||||
import { randomUUID } from 'crypto'
|
||||
|
||||
/**
|
||||
* PgBoss队列适配器
|
||||
*
|
||||
* 适用场景：
|
||||
* - Postgres-Only架构（无需Redis）
|
||||
* - 云原生Serverless环境（SAE）
|
||||
* - 多实例部署需要共享队列
|
||||
* - 关键任务（需要持久化）
|
||||
*
|
||||
* 特点：
|
||||
* - ✅ 无需额外Redis实例，降低成本
|
||||
* - ✅ 多实例自动负载均衡
|
||||
* - ✅ 任务持久化，实例重启不丢失
|
||||
* - ✅ 支持延迟任务、重试、优先级
|
||||
* - ✅ 适合中小规模应用（<10万任务/天）
|
||||
* - ⚠️ 性能低于Redis队列（但足够）
|
||||
*
|
||||
* pg-boss特性：
|
||||
* - 基于Postgres SKIP LOCKED机制
|
||||
* - 自动创建表：platform_schema.job 和 platform_schema.version
|
||||
* - 自动清理过期任务
|
||||
* - 支持CRON定时任务
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const queue = new PgBossQueue(databaseUrl)
|
||||
* await queue.start()
|
||||
*
|
||||
* // 注册处理函数
|
||||
* queue.process('asl:screening', async (job) => {
|
||||
* await processScreening(job.data)
|
||||
* })
|
||||
*
|
||||
* // 创建任务
|
||||
* const job = await queue.push('asl:screening', { projectId: 123 })
|
||||
* ```
|
||||
*/
|
||||
export class PgBossQueue implements JobQueue {
|
||||
private boss: PgBoss
|
||||
private jobs: Map<string, Job> = new Map() // 任务元数据缓存
|
||||
private handlers: Map<string, JobHandler> = new Map()
|
||||
private started: boolean = false
|
||||
|
||||
constructor(connectionString: string, schema: string = 'platform_schema') {
|
||||
this.boss = new PgBoss({
|
||||
connectionString,
|
||||
schema, // 使用platform_schema
|
||||
max: 10, // 最大连接数
|
||||
application_name: 'aiclinical-queue',
|
||||
|
||||
// 调度配置
|
||||
schedule: true, // 启用定时任务
|
||||
|
||||
// 维护配置
|
||||
supervise: true, // 启用监控
|
||||
maintenanceIntervalSeconds: 300, // 每5分钟运行维护任务
|
||||
})
|
||||
|
||||
console.log('[PgBossQueue] Initialized with schema:', schema)
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动队列
|
||||
* 必须在使用前调用
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
if (this.started) return
|
||||
|
||||
try {
|
||||
await this.boss.start()
|
||||
this.started = true
|
||||
console.log('[PgBossQueue] Started successfully')
|
||||
|
||||
// 重新注册所有handler
|
||||
for (const [type, handler] of this.handlers) {
|
||||
await this.registerBossHandler(type, handler)
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[PgBossQueue] Failed to start:', error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 停止队列
|
||||
*/
|
||||
async stop(): Promise<void> {
|
||||
if (!this.started) return
|
||||
|
||||
try {
|
||||
await this.boss.stop()
|
||||
this.started = false
|
||||
console.log('[PgBossQueue] Stopped')
|
||||
} catch (error) {
|
||||
console.error('[PgBossQueue] Failed to stop:', error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 添加任务到队列
|
||||
*
|
||||
* @param type 任务类型
|
||||
* @param data 任务数据
|
||||
* @returns Job对象
|
||||
*/
|
||||
async push<T>(type: string, data: T): Promise<Job<T>> {
|
||||
if (!this.started) {
|
||||
await this.start()
|
||||
}
|
||||
|
||||
try {
|
||||
// 创建任务元数据
|
||||
const jobId = randomUUID()
|
||||
const now = new Date()
|
||||
|
||||
const job: Job<T> = {
|
||||
id: jobId,
|
||||
type,
|
||||
data,
|
||||
status: 'pending',
|
||||
progress: 0,
|
||||
createdAt: now,
|
||||
updatedAt: now
|
||||
}
|
||||
|
||||
// 存储元数据到缓存
|
||||
this.jobs.set(jobId, job)
|
||||
|
||||
// 确保队列存在（幂等操作）
|
||||
try {
|
||||
await this.boss.createQueue(type, {
|
||||
retryLimit: 3,
|
||||
retryDelay: 60,
|
||||
expireInSeconds: 6 * 60 * 60 // 6小时
|
||||
});
|
||||
} catch (error: any) {
|
||||
// 队列已存在时会报错，忽略
|
||||
if (!error.message?.includes('already exists')) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// 发送任务到pg-boss
|
||||
const bossJobId = await this.boss.send(type, {
|
||||
...data,
|
||||
__jobId: jobId, // 嵌入我们的jobId
|
||||
__createdAt: now.toISOString()
|
||||
}, {
|
||||
retryLimit: 3,
|
||||
retryDelay: 60,
|
||||
expireInSeconds: 6 * 60 * 60 // 6小时过期（更适合长批次任务）
|
||||
})
|
||||
|
||||
console.log(`[PgBossQueue] Job pushed: ${jobId} -> pg-boss:${bossJobId} (type: ${type})`)
|
||||
|
||||
return job
|
||||
} catch (error) {
|
||||
console.error(`[PgBossQueue] Failed to push job (type: ${type}):`, error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册任务处理函数
|
||||
*
|
||||
* @param type 任务类型
|
||||
* @param handler 处理函数
|
||||
*/
|
||||
process<T>(type: string, handler: JobHandler<T>): void {
|
||||
this.handlers.set(type, handler)
|
||||
console.log(`[PgBossQueue] Registered handler for job type: ${type}`)
|
||||
|
||||
// 如果已启动，立即注册到pg-boss
|
||||
if (this.started) {
|
||||
this.registerBossHandler(type, handler).catch(err => {
|
||||
console.error(`[PgBossQueue] Failed to register handler for ${type}:`, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册handler到pg-boss
|
||||
* （内部方法）
|
||||
*/
|
||||
private async registerBossHandler<T>(type: string, handler: JobHandler<T>): Promise<void> {
|
||||
// pg-boss 9.x 需要显式创建队列
|
||||
await this.boss.createQueue(type, {
|
||||
retryLimit: 3,
|
||||
retryDelay: 60,
|
||||
expireInSeconds: 6 * 60 * 60 // 6小时
|
||||
});
|
||||
console.log(`[PgBossQueue] Queue created: ${type}`);
|
||||
|
||||
await this.boss.work<Record<string, any>>(type, {
|
||||
batchSize: 1, // 每次处理1个任务
|
||||
pollingIntervalSeconds: 1 // 每秒轮询一次
|
||||
}, async (bossJobs) => {
|
||||
// pg-boss的work handler接收的是Job数组
|
||||
const bossJob = bossJobs[0]
|
||||
if (!bossJob) return
|
||||
|
||||
const { __jobId, __createdAt, ...data } = bossJob.data
|
||||
const jobId = __jobId || randomUUID()
|
||||
|
||||
// 获取或创建Job对象
|
||||
let job = this.jobs.get(jobId)
|
||||
if (!job) {
|
||||
job = {
|
||||
id: jobId,
|
||||
type,
|
||||
data: data as T,
|
||||
status: 'processing',
|
||||
progress: 0,
|
||||
createdAt: new Date(__createdAt || Date.now()),
|
||||
updatedAt: new Date(),
|
||||
startedAt: new Date()
|
||||
}
|
||||
this.jobs.set(jobId, job)
|
||||
} else {
|
||||
job.status = 'processing'
|
||||
job.startedAt = new Date()
|
||||
job.updatedAt = new Date()
|
||||
}
|
||||
|
||||
console.log(`[PgBossQueue] Processing job: ${jobId} (type: ${type})`)
|
||||
|
||||
try {
|
||||
// 执行用户提供的处理函数
|
||||
const result = await handler(job)
|
||||
|
||||
// 标记为完成
|
||||
await this.completeJob(jobId, result)
|
||||
|
||||
return result
|
||||
} catch (error: any) {
|
||||
// 标记为失败
|
||||
await this.failJob(jobId, error.message || String(error))
|
||||
|
||||
// 抛出错误让pg-boss处理重试
|
||||
throw error
|
||||
}
|
||||
})
|
||||
|
||||
console.log(`[PgBossQueue] Handler registered to pg-boss: ${type}`)
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取任务信息
|
||||
*
|
||||
* @param id 任务ID
|
||||
* @returns Job对象或null
|
||||
*/
|
||||
async getJob(id: string): Promise<Job | null> {
|
||||
// 先从缓存查找
|
||||
const cachedJob = this.jobs.get(id)
|
||||
if (cachedJob) {
|
||||
return cachedJob
|
||||
}
|
||||
|
||||
// TODO: 从pg-boss查询（需要额外存储）
|
||||
// 目前只返回缓存中的任务
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新任务进度
|
||||
*
|
||||
* @param id 任务ID
|
||||
* @param progress 进度（0-100）
|
||||
*/
|
||||
async updateProgress(id: string, progress: number): Promise<void> {
|
||||
const job = this.jobs.get(id)
|
||||
if (job) {
|
||||
job.progress = Math.min(100, Math.max(0, progress))
|
||||
job.updatedAt = new Date()
|
||||
this.jobs.set(id, job)
|
||||
|
||||
console.log(`[PgBossQueue] Job progress updated: ${id} -> ${progress}%`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记任务为完成
|
||||
*
|
||||
* @param id 任务ID
|
||||
* @param result 任务结果
|
||||
*/
|
||||
async completeJob(id: string, result: any): Promise<void> {
|
||||
const job = this.jobs.get(id)
|
||||
if (job) {
|
||||
job.status = 'completed'
|
||||
job.progress = 100
|
||||
job.result = result
|
||||
job.completedAt = new Date()
|
||||
job.updatedAt = new Date()
|
||||
this.jobs.set(id, job)
|
||||
|
||||
console.log(`[PgBossQueue] Job completed: ${id} (type: ${job.type})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记任务为失败
|
||||
*
|
||||
* @param id 任务ID
|
||||
* @param error 错误信息
|
||||
*/
|
||||
async failJob(id: string, error: string): Promise<void> {
|
||||
const job = this.jobs.get(id)
|
||||
if (job) {
|
||||
job.status = 'failed'
|
||||
job.error = error
|
||||
job.completedAt = new Date()
|
||||
job.updatedAt = new Date()
|
||||
this.jobs.set(id, job)
|
||||
|
||||
console.error(`[PgBossQueue] Job failed: ${id} (type: ${job.type})`, error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取队列统计信息
|
||||
*/
|
||||
async getStats() {
|
||||
const jobs = Array.from(this.jobs.values())
|
||||
return {
|
||||
total: jobs.length,
|
||||
pending: jobs.filter(j => j.status === 'pending').length,
|
||||
processing: jobs.filter(j => j.status === 'processing').length,
|
||||
completed: jobs.filter(j => j.status === 'completed').length,
|
||||
failed: jobs.filter(j => j.status === 'failed').length
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理已完成的任务（从缓存中）
|
||||
*/
|
||||
cleanup(olderThan: Date = new Date(Date.now() - 24 * 60 * 60 * 1000)) {
|
||||
let removed = 0
|
||||
for (const [id, job] of this.jobs) {
|
||||
if (
|
||||
(job.status === 'completed' || job.status === 'failed') &&
|
||||
job.completedAt &&
|
||||
job.completedAt < olderThan
|
||||
) {
|
||||
this.jobs.delete(id)
|
||||
removed++
|
||||
}
|
||||
}
|
||||
|
||||
if (removed > 0) {
|
||||
console.log(`[PgBossQueue] Cleanup: removed ${removed} old jobs from cache`)
|
||||
}
|
||||
|
||||
return removed
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user