463 lines
15 KiB
TypeScript
463 lines
15 KiB
TypeScript
import type { Prisma } from '@prisma/client'
|
||
import { prisma } from '@/lib/prisma'
|
||
import {
|
||
completeProductPackageUpload,
|
||
type ProductPackageAssetsInput,
|
||
type CompleteProductPackageUploadInput,
|
||
} from '@/lib/product-package-completion'
|
||
|
||
export const TASK_DETAIL_UPLOAD_JOB_STATUS = {
|
||
QUEUED: 'queued',
|
||
PROCESSING: 'processing',
|
||
COMPLETED: 'completed',
|
||
FAILED: 'failed',
|
||
} as const
|
||
|
||
export const TASK_DETAIL_UPLOAD_JOB_STAGE = {
|
||
QUEUED: 'queued',
|
||
VALIDATING: 'validating',
|
||
ASSEMBLING: 'assembling',
|
||
PREVIEWS: 'previews',
|
||
ZIPS: 'zips',
|
||
DATABASE: 'database',
|
||
CLEANUP: 'cleanup',
|
||
COMPLETED: 'completed',
|
||
FAILED: 'failed',
|
||
} as const
|
||
|
||
export type TaskDetailUploadJobStatus =
|
||
(typeof TASK_DETAIL_UPLOAD_JOB_STATUS)[keyof typeof TASK_DETAIL_UPLOAD_JOB_STATUS]
|
||
|
||
export interface CreateTaskDetailUploadJobInput {
|
||
materialId: string
|
||
taskNo?: string | null
|
||
dingtalkApprovalNo?: string | null
|
||
operatorUserId: string
|
||
assets: ProductPackageAssetsInput
|
||
options?: CompleteProductPackageUploadInput['options']
|
||
}
|
||
|
||
export interface TaskDetailUploadJobStatusView {
|
||
jobId: string
|
||
materialId: string
|
||
status: TaskDetailUploadJobStatus | string
|
||
stage: string
|
||
progress: number
|
||
error: string | null
|
||
createdAt: string
|
||
startedAt: string | null
|
||
completedAt: string | null
|
||
}
|
||
|
||
const MAX_CONCURRENT_JOBS = 1
|
||
// 有心跳的正常任务不会触发此门禁；进程被杀后最多等待 30 分钟即可变成
|
||
// failed，而不是把素材永久卡在 processing。超大任务可通过环境变量放宽。
|
||
const DEFAULT_STALE_JOB_MS = 30 * 60 * 1000
|
||
const RECOVERY_BATCH_SIZE = 20
|
||
|
||
let ensureTablePromise: Promise<void> | null = null
|
||
const pendingJobIds = new Set<string>()
|
||
let activeJobCount = 0
|
||
let pumpRunning = false
|
||
|
||
/**
|
||
* 在当前事务内获取任务-详情上传的 advisory lock。
|
||
*
|
||
* pg_advisory_xact_lock() 本身返回 PostgreSQL 的 void 类型。直接用
|
||
* Prisma $queryRaw 读取它会触发 `Failed to deserialize column of type 'void'`，
|
||
* 因此把函数放进子查询，只返回一个普通整数列给 Prisma；锁仍会在当前
|
||
* 事务结束时自动释放。
|
||
*/
|
||
export async function acquireTaskDetailUploadLock(
|
||
tx: Prisma.TransactionClient,
|
||
lockName: string,
|
||
) {
|
||
await tx.$queryRaw`
|
||
SELECT 1 AS locked
|
||
FROM (
|
||
SELECT pg_advisory_xact_lock(hashtext(${lockName}))
|
||
) AS lock_result
|
||
`
|
||
}
|
||
|
||
function getStaleJobMs() {
|
||
const value = Number(process.env.TASK_DETAIL_UPLOAD_STALE_MS)
|
||
return Number.isFinite(value) && value > 0 ? value : DEFAULT_STALE_JOB_MS
|
||
}
|
||
|
||
function toJsonValue(value: unknown): Prisma.InputJsonValue {
|
||
return JSON.parse(JSON.stringify(value)) as Prisma.InputJsonValue
|
||
}
|
||
|
||
function toIso(value: Date | null | undefined) {
|
||
return value ? value.toISOString() : null
|
||
}
|
||
|
||
function toStatusView(job: {
|
||
id: string
|
||
materialId: string
|
||
status: string
|
||
stage: string
|
||
progress: number
|
||
errorMessage: string | null
|
||
createdAt: Date
|
||
startedAt: Date | null
|
||
completedAt: Date | null
|
||
}): TaskDetailUploadJobStatusView {
|
||
return {
|
||
jobId: job.id,
|
||
materialId: job.materialId,
|
||
status: job.status,
|
||
stage: job.stage,
|
||
progress: Math.max(0, Math.min(100, job.progress)),
|
||
error: job.errorMessage || null,
|
||
createdAt: job.createdAt.toISOString(),
|
||
startedAt: toIso(job.startedAt),
|
||
completedAt: toIso(job.completedAt),
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 任务表由 reviewed Prisma migration 创建。请求路径不执行 DDL，避免滚动发布时
|
||
* 多个容器同时抢迁移锁；若迁移未执行，调用方会收到明确的数据库错误并阻止入队。
|
||
*/
|
||
export async function ensureProductPackageUploadJobTable() {
|
||
if (!ensureTablePromise) {
|
||
ensureTablePromise = Promise.resolve()
|
||
}
|
||
return ensureTablePromise
|
||
}
|
||
|
||
async function updateJobProgress(
|
||
jobId: string,
|
||
update: { stage: string; progress: number; message?: string },
|
||
) {
|
||
const progress = Math.max(0, Math.min(99, Math.round(update.progress)))
|
||
await prisma.productPackageUploadJob.updateMany({
|
||
where: { id: jobId, status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING },
|
||
data: {
|
||
stage: update.stage,
|
||
progress,
|
||
...(update.message ? { errorMessage: update.message } : {}),
|
||
},
|
||
})
|
||
}
|
||
|
||
async function claimTaskDetailUploadJob(jobId: string) {
|
||
return prisma.$transaction(async (tx) => {
|
||
// 用事务级 advisory lock 串行化所有容器的“领取任务”动作，保证全局最多一个
|
||
// 任务进入 Sharp/ZIP 重处理，而不是只在单个 Node 进程内限流。
|
||
await acquireTaskDetailUploadLock(tx, 'product-package-upload-worker')
|
||
const activeCount = await tx.productPackageUploadJob.count({
|
||
where: { status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING },
|
||
})
|
||
if (activeCount > 0) return false
|
||
|
||
const claimed = await tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
},
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.VALIDATING,
|
||
progress: 5,
|
||
startedAt: new Date(),
|
||
completedAt: null,
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
return claimed.count === 1
|
||
})
|
||
}
|
||
|
||
async function reconcileCommittedTaskDetailUploadJob(jobId: string) {
|
||
const job = await prisma.productPackageUploadJob.findUnique({
|
||
where: { id: jobId },
|
||
select: {
|
||
id: true,
|
||
status: true,
|
||
materialId: true,
|
||
material: { select: { zipFilePath: true, taskStatus: true } },
|
||
},
|
||
})
|
||
if (!job || (job.status !== TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED && job.status !== TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING)) {
|
||
return !job
|
||
}
|
||
|
||
const assetCount = await prisma.materialAsset.count({ where: { materialId: job.materialId } })
|
||
if (job.material.taskStatus !== 'completed' || !job.material.zipFilePath || assetCount === 0) return false
|
||
|
||
// 主资产事务已提交但容器可能在 job.completed 写入前被杀掉：不要拿已清理的
|
||
// upload session 重跑，直接把持久化任务补记为完成。
|
||
await prisma.productPackageUploadJob.updateMany({
|
||
where: { id: jobId, status: { in: [TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED, TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING] } },
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.COMPLETED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.COMPLETED,
|
||
progress: 100,
|
||
completedAt: new Date(),
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
return true
|
||
}
|
||
|
||
async function processOneTaskDetailUploadJob(jobId: string) {
|
||
if (await reconcileCommittedTaskDetailUploadJob(jobId)) return
|
||
const claimed = await claimTaskDetailUploadJob(jobId)
|
||
if (!claimed) {
|
||
const pendingJob = await prisma.productPackageUploadJob.findUnique({
|
||
where: { id: jobId },
|
||
select: { status: true },
|
||
})
|
||
if (pendingJob?.status === TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED) {
|
||
setTimeout(() => triggerTaskDetailUploadJob(jobId), 5_000)
|
||
}
|
||
return
|
||
}
|
||
|
||
const job = await prisma.productPackageUploadJob.findUnique({ where: { id: jobId } })
|
||
if (!job) {
|
||
return
|
||
}
|
||
|
||
const heartbeatTimer = setInterval(() => {
|
||
void prisma.productPackageUploadJob.updateMany({
|
||
where: { id: jobId, status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING },
|
||
data: { updatedAt: new Date() },
|
||
}).catch((heartbeatError) => {
|
||
console.warn(`[TaskDetailUploadJob] heartbeat failed job=${jobId}:`, heartbeatError)
|
||
})
|
||
}, 30_000)
|
||
|
||
try {
|
||
try {
|
||
await completeProductPackageUpload({
|
||
materialId: job.materialId,
|
||
taskNo: job.taskNo,
|
||
dingtalkApprovalNo: job.dingtalkApprovalNo,
|
||
operatorUserId: job.operatorUserId,
|
||
assets: job.assets as ProductPackageAssetsInput,
|
||
options: {
|
||
...(job.options && typeof job.options === 'object' ? job.options as Record<string, unknown> : {}),
|
||
preserveMaterialOnError: true,
|
||
onProgress: (update) => updateJobProgress(jobId, update),
|
||
},
|
||
})
|
||
} catch (error) {
|
||
const errorMessage = error instanceof Error ? error.message : '任务-详情上传处理失败'
|
||
console.error(`[TaskDetailUploadJob] failed job=${jobId}:`, error)
|
||
try {
|
||
await prisma.productPackageUploadJob.updateMany({
|
||
where: { id: jobId },
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.FAILED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.FAILED,
|
||
errorMessage: errorMessage.slice(0, 2000),
|
||
completedAt: new Date(),
|
||
},
|
||
})
|
||
} catch (jobFailureRecordError) {
|
||
console.error(`[TaskDetailUploadJob] failed record write failed job=${jobId}:`, jobFailureRecordError)
|
||
}
|
||
await prisma.material.updateMany({
|
||
where: { id: job.materialId, taskStatus: { not: 'completed' } },
|
||
data: { taskStatus: 'failed' },
|
||
}).catch((statusError) => {
|
||
console.error(`[TaskDetailUploadJob] failed to mark material status job=${jobId}:`, statusError)
|
||
})
|
||
return
|
||
}
|
||
|
||
try {
|
||
await prisma.productPackageUploadJob.updateMany({
|
||
where: { id: jobId },
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.COMPLETED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.COMPLETED,
|
||
progress: 100,
|
||
completedAt: new Date(),
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
} catch (completionRecordError) {
|
||
// 素材处理已经成功时，不能把 job 反写成 failed。保留 processing，交给
|
||
// reconcile/周期恢复在数据库恢复后补记 completed。
|
||
console.error(`[TaskDetailUploadJob] completed record write failed job=${jobId}:`, completionRecordError)
|
||
try {
|
||
await reconcileCommittedTaskDetailUploadJob(jobId)
|
||
} catch (reconcileError) {
|
||
console.error(`[TaskDetailUploadJob] completed record reconcile failed job=${jobId}:`, reconcileError)
|
||
}
|
||
}
|
||
} finally {
|
||
clearInterval(heartbeatTimer)
|
||
}
|
||
}
|
||
|
||
async function pumpTaskDetailUploadJobs() {
|
||
if (pumpRunning) {
|
||
return
|
||
}
|
||
|
||
pumpRunning = true
|
||
try {
|
||
while (activeJobCount < MAX_CONCURRENT_JOBS && pendingJobIds.size > 0) {
|
||
const nextJobId = pendingJobIds.values().next().value as string | undefined
|
||
if (!nextJobId) {
|
||
break
|
||
}
|
||
pendingJobIds.delete(nextJobId)
|
||
activeJobCount += 1
|
||
void processOneTaskDetailUploadJob(nextJobId)
|
||
.catch((error) => {
|
||
console.error(`[TaskDetailUploadJob] worker crashed job=${nextJobId}:`, error)
|
||
})
|
||
.finally(() => {
|
||
activeJobCount -= 1
|
||
void pumpTaskDetailUploadJobs()
|
||
})
|
||
}
|
||
} finally {
|
||
pumpRunning = false
|
||
}
|
||
}
|
||
|
||
export function triggerTaskDetailUploadJob(jobId: string) {
|
||
pendingJobIds.add(jobId)
|
||
void pumpTaskDetailUploadJobs().catch((error) => {
|
||
console.error('[TaskDetailUploadJob] queue pump failed:', error)
|
||
})
|
||
}
|
||
|
||
export async function createTaskDetailUploadJob(input: CreateTaskDetailUploadJobInput) {
|
||
await ensureProductPackageUploadJobTable()
|
||
const job = await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, 'product-package-upload-create')
|
||
const existing = await tx.productPackageUploadJob.findFirst({
|
||
where: {
|
||
materialId: input.materialId,
|
||
operatorUserId: input.operatorUserId,
|
||
status: {
|
||
in: [
|
||
TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
TASK_DETAIL_UPLOAD_JOB_STATUS.COMPLETED,
|
||
],
|
||
},
|
||
},
|
||
orderBy: { createdAt: 'desc' },
|
||
})
|
||
if (existing) return existing
|
||
|
||
const created = await tx.productPackageUploadJob.create({
|
||
data: {
|
||
materialId: input.materialId,
|
||
operatorUserId: input.operatorUserId,
|
||
taskNo: input.taskNo?.trim() || null,
|
||
dingtalkApprovalNo: input.dingtalkApprovalNo?.trim() || null,
|
||
assets: toJsonValue(input.assets),
|
||
options: input.options ? toJsonValue(input.options) : undefined,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.QUEUED,
|
||
progress: 0,
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
|
||
// 素材列表/详情页可以据此识别“正在后台处理”的空素材，用户关闭上传页后
|
||
// 仍然能知道它不是一个普通的空卡片。完成函数会把状态推进到 completed。
|
||
await tx.material.update({
|
||
where: { id: input.materialId },
|
||
data: { taskStatus: 'uploading' },
|
||
})
|
||
|
||
return created
|
||
})
|
||
|
||
triggerTaskDetailUploadJob(job.id)
|
||
return toStatusView(job)
|
||
}
|
||
|
||
export async function getTaskDetailUploadJobStatus(jobId: string, operatorUserId: string) {
|
||
await ensureProductPackageUploadJobTable()
|
||
const job = await prisma.productPackageUploadJob.findFirst({
|
||
where: {
|
||
id: jobId,
|
||
operatorUserId,
|
||
},
|
||
})
|
||
|
||
return job ? toStatusView(job) : null
|
||
}
|
||
|
||
export async function getLatestTaskDetailUploadJobStatus(materialId: string, operatorUserId: string) {
|
||
await ensureProductPackageUploadJobTable()
|
||
const job = await prisma.productPackageUploadJob.findFirst({
|
||
where: {
|
||
materialId,
|
||
operatorUserId,
|
||
},
|
||
orderBy: { createdAt: 'desc' },
|
||
})
|
||
|
||
return job ? toStatusView(job) : null
|
||
}
|
||
|
||
/**
|
||
* 服务重启后恢复未完成队列。processing 会在心跳停止并超过 stale 窗口后标记失败，
|
||
* 不直接重跑可能已经写入部分资产的非幂等处理链路。
|
||
*/
|
||
export async function recoverTaskDetailUploadJobs() {
|
||
await ensureProductPackageUploadJobTable()
|
||
const staleBefore = new Date(Date.now() - getStaleJobMs())
|
||
|
||
const staleJobs = await prisma.productPackageUploadJob.findMany({
|
||
where: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
updatedAt: { lt: staleBefore },
|
||
},
|
||
select: { id: true, materialId: true },
|
||
})
|
||
|
||
// 处理链路会写入 ZIP、预览图和资产记录，无法安全地把中断中的任务直接
|
||
// 重跑，否则可能重复资产或覆盖仍被数据库引用的文件。只有确认完成标记已
|
||
// 落库的任务才补记 completed，其余任务标记 failed，交给用户重新提交。
|
||
for (const staleJob of staleJobs) {
|
||
if (await reconcileCommittedTaskDetailUploadJob(staleJob.id)) continue
|
||
|
||
await prisma.productPackageUploadJob.updateMany({
|
||
where: { id: staleJob.id, status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING },
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.FAILED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.FAILED,
|
||
errorMessage: '后台处理进程中断，请重新提交上传',
|
||
completedAt: new Date(),
|
||
},
|
||
})
|
||
await prisma.material.updateMany({
|
||
where: { id: staleJob.materialId, taskStatus: { not: 'completed' } },
|
||
data: { taskStatus: 'failed' },
|
||
}).catch((statusError) => {
|
||
console.error(`[TaskDetailUploadJob] failed to mark stale material=${staleJob.materialId}:`, statusError)
|
||
})
|
||
}
|
||
|
||
const queuedJobs = await prisma.productPackageUploadJob.findMany({
|
||
where: { status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED },
|
||
orderBy: { createdAt: 'asc' },
|
||
take: RECOVERY_BATCH_SIZE,
|
||
select: { id: true },
|
||
})
|
||
|
||
for (const job of queuedJobs) {
|
||
triggerTaskDetailUploadJob(job.id)
|
||
}
|
||
|
||
if (queuedJobs.length > 0) {
|
||
console.log(`[TaskDetailUploadJob] recovered ${queuedJobs.length} queued job(s)`)
|
||
}
|
||
return queuedJobs.length
|
||
}
|