216 lines
18 KiB
TypeScript
216 lines
18 KiB
TypeScript
import { createHash, randomUUID } from 'crypto'
|
|
import type { Prisma } from '@prisma/client'
|
|
import { prisma } from '@/lib/prisma'
|
|
import { DetailGeneratorError } from './access'
|
|
import type { DetailAsset, DetailAssetKind, DetailDocument, DetailJob, DetailJobKind, DetailJobStatus, DetailProject } from './types'
|
|
|
|
type Database = Prisma.TransactionClient | typeof prisma
|
|
type Row = Record<string, any>
|
|
export interface StoredDetailAsset extends DetailAsset { byteSize: number }
|
|
export interface StoredDetailJob extends DetailJob {
|
|
fingerprint: string; projectRevision: number; idempotencyKey: string; leaseToken: string | null; leaseExpiresAt: Date | null; progress: number
|
|
}
|
|
const projectColumns = `id, owner_id AS "ownerId", document, revision, created_at AS "createdAt", updated_at AS "updatedAt"`
|
|
const assetColumns = `a.id, a.project_id AS "projectId", a.owner_id AS "ownerId", a.kind, a.name, a.mime_type AS "mimeType", a.storage_key AS "storageKey", a.byte_size AS "byteSize", a.width, a.height, a.extracted_text AS "extractedText", a.created_at AS "createdAt"`
|
|
const jobColumns = `id, project_id AS "projectId", owner_id AS "ownerId", kind, status, payload, checkpoint, fingerprint, project_revision AS "projectRevision", idempotency_key AS "idempotencyKey", lease_token AS "leaseToken", lease_expires_at AS "leaseExpiresAt", progress, error, request_id AS "requestId", created_at AS "createdAt", updated_at AS "updatedAt"`
|
|
function iso(value: Date | string): string { return value instanceof Date ? value.toISOString() : value }
|
|
function project(row: Row): DetailProject { return { ...row, createdAt: iso(row.createdAt), updatedAt: iso(row.updatedAt) } as DetailProject }
|
|
function asset(row: Row): StoredDetailAsset { return { ...row, createdAt: iso(row.createdAt) } as StoredDetailAsset }
|
|
function job(row: Row): StoredDetailJob { return { ...row, createdAt: iso(row.createdAt), updatedAt: iso(row.updatedAt) } as StoredDetailJob }
|
|
|
|
export async function detailTransactionLock(tx: Database, key: string): Promise<void> {
|
|
await tx.$queryRawUnsafe('SELECT 1 AS locked FROM (SELECT pg_advisory_xact_lock(hashtext($1))) AS lock_result', key)
|
|
}
|
|
|
|
export async function listDetailProjects(ownerId: string): Promise<DetailProject[]> {
|
|
const rows = await prisma.$queryRawUnsafe<Row[]>(`SELECT ${projectColumns} FROM detail_generator_projects WHERE owner_id = $1 AND deleted_at IS NULL ORDER BY updated_at DESC LIMIT 100`, ownerId)
|
|
return rows.map(project)
|
|
}
|
|
|
|
export async function findDetailProject(id: string, ownerId: string, db: Database = prisma): Promise<DetailProject> {
|
|
const rows = await db.$queryRawUnsafe<Row[]>(`SELECT ${projectColumns} FROM detail_generator_projects WHERE id = $1 AND owner_id = $2 AND deleted_at IS NULL`, id, ownerId)
|
|
if (!rows[0]) throw new DetailGeneratorError('项目不存在', 404, 'not_found')
|
|
return project(rows[0])
|
|
}
|
|
|
|
export async function createDetailProject(ownerId: string, document: DetailDocument): Promise<DetailProject> {
|
|
const rows = await prisma.$queryRawUnsafe<Row[]>(`INSERT INTO detail_generator_projects (id, owner_id, document) VALUES ($1, $2, $3::jsonb) RETURNING ${projectColumns}`, randomUUID(), ownerId, JSON.stringify(document))
|
|
return project(rows[0])
|
|
}
|
|
|
|
export async function assertDetailProjectIdle(id: string, ownerId: string, db: Database = prisma): Promise<void> {
|
|
const rows = await db.$queryRawUnsafe<Row[]>('SELECT id FROM detail_generator_jobs WHERE project_id = $1 AND owner_id = $2 AND status IN (\'queued\', \'running\') LIMIT 1', id, ownerId)
|
|
if (rows.length) throw new DetailGeneratorError('项目任务正在执行，请完成后再编辑', 409, 'project_busy')
|
|
}
|
|
|
|
export async function updateDetailProject(id: string, ownerId: string, revision: number, document: DetailDocument): Promise<DetailProject> {
|
|
return prisma.$transaction(async tx => {
|
|
await detailTransactionLock(tx, `detail-project:${id}`)
|
|
await findDetailProject(id, ownerId, tx)
|
|
await assertDetailProjectIdle(id, ownerId, tx)
|
|
return updateDetailProjectRevision(tx, id, ownerId, revision, document)
|
|
})
|
|
}
|
|
|
|
async function updateDetailProjectRevision(db: Database, id: string, ownerId: string, revision: number, document: DetailDocument): Promise<DetailProject> {
|
|
const rows = await db.$queryRawUnsafe<Row[]>(`UPDATE detail_generator_projects SET document = $4::jsonb, revision = revision + 1, updated_at = NOW() WHERE id = $1 AND owner_id = $2 AND revision = $3 AND deleted_at IS NULL RETURNING ${projectColumns}`, id, ownerId, revision, JSON.stringify(document))
|
|
if (!rows[0]) throw new DetailGeneratorError('项目已更新，请刷新后重试', 409, 'revision_conflict')
|
|
return project(rows[0])
|
|
}
|
|
|
|
export async function deleteDetailProject(id: string, ownerId: string): Promise<void> {
|
|
await prisma.$transaction(async tx => {
|
|
await detailTransactionLock(tx, `detail-project:${id}`)
|
|
await findDetailProject(id, ownerId, tx)
|
|
await assertDetailProjectIdle(id, ownerId, tx)
|
|
await tx.$executeRawUnsafe('UPDATE detail_generator_projects SET deleted_at = NOW(), updated_at = NOW() WHERE id = $1 AND owner_id = $2', id, ownerId)
|
|
})
|
|
}
|
|
|
|
export async function listDetailAssets(projectId: string, ownerId: string): Promise<StoredDetailAsset[]> {
|
|
await findDetailProject(projectId, ownerId)
|
|
const rows = await prisma.$queryRawUnsafe<Row[]>(`SELECT ${assetColumns} FROM detail_generator_assets a WHERE a.project_id = $1 AND a.owner_id = $2 ORDER BY a.created_at`, projectId, ownerId)
|
|
return rows.map(asset)
|
|
}
|
|
|
|
export async function findDetailAsset(id: string, ownerId: string): Promise<StoredDetailAsset> {
|
|
const rows = await prisma.$queryRawUnsafe<Row[]>(`SELECT ${assetColumns} FROM detail_generator_assets a JOIN detail_generator_projects p ON p.id = a.project_id AND p.owner_id = a.owner_id WHERE a.id = $1 AND a.owner_id = $2 AND p.deleted_at IS NULL`, id, ownerId)
|
|
if (!rows[0]) throw new DetailGeneratorError('文件不存在', 404, 'not_found')
|
|
return asset(rows[0])
|
|
}
|
|
|
|
export async function insertDetailAsset(input: { id?: string; projectId: string; ownerId: string; kind: DetailAssetKind; name: string; mimeType: string; storageKey: string; byteSize: number; width?: number | null; height?: number | null; extractedText?: string | null }, db: Database = prisma): Promise<StoredDetailAsset> {
|
|
await findDetailProject(input.projectId, input.ownerId, db)
|
|
const rows = await db.$queryRawUnsafe<Row[]>(`INSERT INTO detail_generator_assets AS a (id, project_id, owner_id, kind, name, mime_type, storage_key, byte_size, width, height, extracted_text) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) RETURNING ${assetColumns}`, input.id || randomUUID(), input.projectId, input.ownerId, input.kind, input.name, input.mimeType, input.storageKey, input.byteSize, input.width ?? null, input.height ?? null, input.extractedText ?? null)
|
|
return asset(rows[0])
|
|
}
|
|
|
|
export async function assertDetailDocumentAssets(document: DetailDocument, projectId: string, ownerId: string): Promise<void> {
|
|
const required = [
|
|
...document.product.assetIds.map(id => ({ id, kinds: ['product'] })),
|
|
...document.product.referenceAssetIds.map(id => ({ id, kinds: ['reference'] })),
|
|
...document.product.documentAssetIds.map(id => ({ id, kinds: ['document'] })),
|
|
...(document.brand.logoAssetId ? [{ id: document.brand.logoAssetId, kinds: ['logo'] }] : []),
|
|
...document.screens.flatMap(screen => [...(screen.image ? [{ id: screen.image.assetId, kinds: ['generated', 'product', 'reference'] }] : []), ...(screen.logo ? [{ id: screen.logo.assetId, kinds: ['logo'] }] : [])]),
|
|
]
|
|
if (!required.length) return
|
|
const assets = await listDetailAssets(projectId, ownerId)
|
|
const map = new Map(assets.map(value => [value.id, value]))
|
|
for (const entry of required) {
|
|
const found = map.get(entry.id)
|
|
if (!found || found.projectId !== projectId || !entry.kinds.includes(found.kind)) throw new DetailGeneratorError('项目引用了不可访问或类型不匹配的文件', 403, 'invalid_asset')
|
|
}
|
|
}
|
|
|
|
export async function attachDetailAsset(input: Parameters<typeof insertDetailAsset>[0], revision: number, document: DetailDocument): Promise<{ asset: StoredDetailAsset; project: DetailProject }> {
|
|
return prisma.$transaction(async tx => {
|
|
await detailTransactionLock(tx, `detail-project:${input.projectId}`)
|
|
await findDetailProject(input.projectId, input.ownerId, tx)
|
|
await assertDetailProjectIdle(input.projectId, input.ownerId, tx)
|
|
const stored = await insertDetailAsset(input, tx)
|
|
const updated = await updateDetailProjectRevision(tx, input.projectId, input.ownerId, revision, document)
|
|
return { asset: stored, project: updated }
|
|
})
|
|
}
|
|
|
|
export async function listDetailJobs(projectId: string, ownerId: string): Promise<StoredDetailJob[]> {
|
|
await findDetailProject(projectId, ownerId)
|
|
const rows = await prisma.$queryRawUnsafe<Row[]>(`SELECT ${jobColumns} FROM detail_generator_jobs WHERE project_id = $1 AND owner_id = $2 ORDER BY created_at DESC LIMIT 50`, projectId, ownerId)
|
|
return rows.map(job)
|
|
}
|
|
|
|
export async function findDetailJob(id: string, ownerId: string): Promise<StoredDetailJob> {
|
|
const rows = await prisma.$queryRawUnsafe<Row[]>(`SELECT ${jobColumns} FROM detail_generator_jobs WHERE id = $1 AND owner_id = $2 AND EXISTS (SELECT 1 FROM detail_generator_projects p WHERE p.id = project_id AND p.owner_id = $2 AND p.deleted_at IS NULL)`, id, ownerId)
|
|
if (!rows[0]) throw new DetailGeneratorError('任务不存在', 404, 'not_found')
|
|
return job(rows[0])
|
|
}
|
|
|
|
export async function enqueueDetailJob(input: { projectId: string; ownerId: string; kind: DetailJobKind; idempotencyKey: string; payload?: Record<string, unknown>; revision: number }): Promise<StoredDetailJob> {
|
|
const payload = input.payload || {}
|
|
const fingerprint = createHash('sha256').update(JSON.stringify([input.projectId, input.kind, payload])).digest('hex')
|
|
return prisma.$transaction(async tx => {
|
|
await detailTransactionLock(tx, 'detail-generator-admission')
|
|
await detailTransactionLock(tx, `detail-project:${input.projectId}`)
|
|
const previous = await tx.$queryRawUnsafe<Row[]>(`SELECT ${jobColumns} FROM detail_generator_jobs WHERE owner_id = $1 AND idempotency_key = $2`, input.ownerId, input.idempotencyKey)
|
|
if (previous[0]) {
|
|
if (previous[0].fingerprint !== fingerprint) throw new DetailGeneratorError('同一请求编号不能用于不同操作', 409, 'idempotency_conflict')
|
|
await findDetailProject(input.projectId, input.ownerId, tx)
|
|
return job(previous[0])
|
|
}
|
|
const current = await findDetailProject(input.projectId, input.ownerId, tx)
|
|
if (current.revision !== input.revision) throw new DetailGeneratorError('项目已更新，请刷新后重试', 409, 'revision_conflict')
|
|
const busy = await tx.$queryRawUnsafe<Row[]>('SELECT id FROM detail_generator_jobs WHERE owner_id = $1 AND status IN (\'queued\', \'running\') LIMIT 1', input.ownerId)
|
|
if (busy.length) throw new DetailGeneratorError('您已有任务正在执行，请完成后再提交', 429, 'user_capacity')
|
|
if (input.kind !== 'export') {
|
|
const unresolved = await tx.$queryRawUnsafe<Row[]>('SELECT id FROM detail_generator_jobs WHERE project_id = $1 AND owner_id = $2 AND status = \'unknown\' LIMIT 1', input.projectId, input.ownerId)
|
|
if (unresolved.length) throw new DetailGeneratorError('该项目存在结果未知的模型请求，请先核查服务商记录', 409, 'outcome_unknown')
|
|
}
|
|
const rows = await tx.$queryRawUnsafe<Row[]>(`INSERT INTO detail_generator_jobs (id, project_id, owner_id, kind, idempotency_key, fingerprint, project_revision, payload) VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb) RETURNING ${jobColumns}`, randomUUID(), input.projectId, input.ownerId, input.kind, input.idempotencyKey, fingerprint, current.revision, JSON.stringify({ ...payload, document: current.document }))
|
|
return job(rows[0])
|
|
})
|
|
}
|
|
|
|
/** Execution admission is serialized across worker processes using the database clock. */
|
|
export async function claimDetailJob(): Promise<StoredDetailJob | null> {
|
|
return prisma.$transaction(async tx => {
|
|
await detailTransactionLock(tx, 'detail-generator-execution')
|
|
await tx.$executeRawUnsafe(`UPDATE detail_generator_jobs SET status = CASE WHEN checkpoint->>'dispatchPending' = 'true' THEN 'unknown' ELSE 'queued' END, error = CASE WHEN checkpoint->>'dispatchPending' = 'true' THEN '执行中断，模型请求结果未知；请核查服务商记录，勿重复提交' ELSE NULL END, lease_token = NULL, lease_expires_at = NULL, updated_at = NOW() WHERE status = 'running' AND lease_expires_at < NOW()`)
|
|
const count = await tx.$queryRawUnsafe<Row[]>('SELECT COUNT(*)::int AS count FROM detail_generator_jobs WHERE status = \'running\' AND lease_expires_at >= NOW()')
|
|
if (Number(count[0]?.count) >= 2) return null
|
|
const candidates = await tx.$queryRawUnsafe<Row[]>(`SELECT ${jobColumns} FROM detail_generator_jobs j WHERE status = 'queued' AND EXISTS (SELECT 1 FROM detail_generator_projects p WHERE p.id = j.project_id AND p.deleted_at IS NULL) AND NOT EXISTS (SELECT 1 FROM detail_generator_jobs other WHERE other.owner_id = j.owner_id AND other.status = 'running' AND other.lease_expires_at >= NOW()) ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED`)
|
|
if (!candidates[0]) return null
|
|
const rows = await tx.$queryRawUnsafe<Row[]>(`UPDATE detail_generator_jobs SET status = 'running', lease_token = $2, lease_expires_at = NOW() + INTERVAL '10 minutes', started_at = COALESCE(started_at, NOW()), updated_at = NOW() WHERE id = $1 RETURNING ${jobColumns}`, candidates[0].id, randomUUID())
|
|
return job(rows[0])
|
|
})
|
|
}
|
|
|
|
export async function renewDetailJobLease(id: string, token: string): Promise<boolean> {
|
|
const count = await prisma.$executeRawUnsafe(`UPDATE detail_generator_jobs SET lease_expires_at = NOW() + INTERVAL '10 minutes', updated_at = NOW() WHERE id = $1 AND lease_token = $2 AND status = 'running' AND lease_expires_at >= NOW()`, id, token)
|
|
return count === 1
|
|
}
|
|
|
|
export async function checkpointDetailJob(id: string, token: string, checkpoint: Record<string, unknown>, progress: number): Promise<void> {
|
|
const count = await prisma.$executeRawUnsafe(`UPDATE detail_generator_jobs SET checkpoint = $3::jsonb, progress = $4, updated_at = NOW() WHERE id = $1 AND lease_token = $2 AND status = 'running' AND lease_expires_at >= NOW()`, id, token, JSON.stringify(checkpoint), progress)
|
|
if (count !== 1) throw new DetailGeneratorError('任务执行租约已失效', 409, 'lease_lost')
|
|
}
|
|
|
|
export async function saveDetailJobDocument(jobId: string, token: string, document: DetailDocument, checkpoint: Record<string, unknown>, progress: number): Promise<DetailProject> {
|
|
return prisma.$transaction(async tx => {
|
|
const rows = await tx.$queryRawUnsafe<Row[]>(`SELECT ${jobColumns} FROM detail_generator_jobs WHERE id = $1 AND lease_token = $2 AND status = 'running' AND lease_expires_at >= NOW() FOR UPDATE`, jobId, token)
|
|
if (!rows[0]) throw new DetailGeneratorError('任务执行租约已失效', 409, 'lease_lost')
|
|
const current = job(rows[0])
|
|
const updated = await updateDetailProjectRevision(tx, current.projectId, current.ownerId, current.projectRevision, document)
|
|
await tx.$executeRawUnsafe(`UPDATE detail_generator_jobs SET project_revision = $3, checkpoint = $4::jsonb, progress = $5, updated_at = NOW() WHERE id = $1 AND lease_token = $2`, jobId, token, updated.revision, JSON.stringify(checkpoint), progress)
|
|
return updated
|
|
})
|
|
}
|
|
|
|
export async function saveDetailJobAsset(jobId: string, token: string, input: Parameters<typeof insertDetailAsset>[0], document: DetailDocument | null, checkpoint: Record<string, unknown>, progress: number): Promise<StoredDetailAsset> {
|
|
return prisma.$transaction(async tx => {
|
|
const rows = await tx.$queryRawUnsafe<Row[]>(`SELECT ${jobColumns} FROM detail_generator_jobs WHERE id = $1 AND lease_token = $2 AND status = 'running' AND lease_expires_at >= NOW() FOR UPDATE`, jobId, token)
|
|
if (!rows[0]) throw new DetailGeneratorError('任务执行租约已失效', 409, 'lease_lost')
|
|
const current = job(rows[0])
|
|
if (current.projectId !== input.projectId || current.ownerId !== input.ownerId) throw new DetailGeneratorError('任务文件归属无效', 403)
|
|
const stored = await insertDetailAsset(input, tx)
|
|
const revision = document ? (await updateDetailProjectRevision(tx, current.projectId, current.ownerId, current.projectRevision, document)).revision : current.projectRevision
|
|
await tx.$executeRawUnsafe(`UPDATE detail_generator_jobs SET project_revision = $3, checkpoint = $4::jsonb, progress = $5, updated_at = NOW() WHERE id = $1 AND lease_token = $2`, jobId, token, revision, JSON.stringify(checkpoint), progress)
|
|
return stored
|
|
})
|
|
}
|
|
|
|
export async function finishDetailJob(id: string, token: string, status: DetailJobStatus, checkpoint: Record<string, unknown>, error: string | null = null, requestId: string | null = null): Promise<void> {
|
|
await prisma.$executeRawUnsafe(`UPDATE detail_generator_jobs SET status = $3, checkpoint = $4::jsonb, error = $5, request_id = $6, progress = CASE WHEN $3 = 'succeeded' THEN 100 ELSE progress END, lease_token = NULL, lease_expires_at = NULL, completed_at = NOW(), updated_at = NOW() WHERE id = $1 AND lease_token = $2 AND status = 'running' AND lease_expires_at >= NOW()`, id, token, status, JSON.stringify(checkpoint), error, requestId)
|
|
}
|
|
|
|
export function detailPublicAsset(value: StoredDetailAsset) {
|
|
const { storageKey: _storageKey, extractedText: _extractedText, ...safe } = value
|
|
return { ...safe, url: `/api/detail-generator/assets/${value.id}` }
|
|
}
|
|
export function detailPublicJob(value: StoredDetailJob) {
|
|
const format = value.payload.format
|
|
const payload = value.kind === 'export' && typeof format === 'string' && ['png', 'jpg', 'long-png', 'psd'].includes(format) ? { format } : {}
|
|
return { id: value.id, projectId: value.projectId, ownerId: value.ownerId, kind: value.kind, status: value.status, progress: value.progress,
|
|
checkpoint: { stage: value.checkpoint.stage, exportAssetId: value.checkpoint.exportAssetId, completedScreenIds: value.checkpoint.completedScreenIds },
|
|
payload, error: value.error, requestId: value.requestId, createdAt: value.createdAt, updatedAt: value.updatedAt }
|
|
}
|