import { randomUUID } from 'node:crypto'; import { Inject, Injectable } from '@nestjs/common'; import { createImageSuiteSchema, imageSuitePlanErrorMessages, createMediaAiEditRequestSchema, type CreateImageSuiteRequest, type ImageSuite, type ImageSuiteInput, type ImageSuitePlanInput, type ImageSuiteShot, type ListProductImageTasksQuery, } from '@atlas/contracts'; import type { Database, DatabaseTransaction } from '@atlas/database'; import { ApiError } from '../common/api-error.js'; import { IdempotencyService } from '../common/idempotency.service.js'; import { DATABASE } from '../database/database.module.js'; import { MediaAiEditService } from './media-ai-edit.service.js'; type SuiteRow = Omit< ImageSuite, 'items' | 'operation' | 'created_at' | 'completed_at' > & { created_at: Date | string; completed_at: Date | string | null }; const iso = (value: Date | string) => value instanceof Date ? value.toISOString() : value; @Injectable() export class ImageSuiteService { public constructor( @Inject(DATABASE) private readonly database: Database, @Inject(IdempotencyService) private readonly idempotency: IdempotencyService, @Inject(MediaAiEditService) private readonly edits: MediaAiEditService, ) {} public async create( workspaceId: string, userId: string, productId: string, key: string, request: ImageSuiteInput | ImageSuitePlanInput | CreateImageSuiteRequest, purpose: 'plan' | 'generate', ) { const path = `/api/products/${productId}/ai/${purpose === 'plan' ? 'suite-plans' : 'image-suites'}`; return this.idempotency.run( { workspaceId, userId, method: 'POST', path, key, request }, async (tx) => { const variant = await tx.query<{ id: string }>( 'SELECT id FROM atlas.product_variants WHERE id = $1 AND product_id = $2 AND deleted_at IS NULL', [request.variant_id, productId], ); if (!variant.rows[0]) throw new ApiError( 404, 'RESOURCE_NOT_FOUND', '目标 SKU 不存在或不属于此商品,请先保存新 SKU', ); const assets = await tx.query<{ id: string }>( `SELECT id FROM atlas.media_assets WHERE id = ANY($1::uuid[]) AND deleted_at IS NULL AND content_type IN ('image/jpeg', 'image/png', 'image/webp') AND byte_size <= 20971520 AND width_px BETWEEN 1 AND 10000 AND height_px BETWEEN 1 AND 10000`, [request.reference_asset_ids], ); if (assets.rows.length !== request.reference_asset_ids.length) throw new ApiError( 404, 'RESOURCE_NOT_FOUND', '参考素材不存在或不可处理', ); const shots = purpose === 'generate' ? createImageSuiteSchema.parse(request).shots : []; const { shots: _shots, ...input } = request as CreateImageSuiteRequest; const created = await tx.query( `INSERT INTO atlas.product_image_suites (workspace_id, product_id, variant_id, purpose, input, shots, created_by_user_id) VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4::jsonb, $5::jsonb, $6) RETURNING *`, [ productId, input.variant_id, purpose, JSON.stringify(input), JSON.stringify(shots), userId, ], ); const row = created.rows[0]; for (const [ position, assetId, ] of input.reference_asset_ids.entries()) await tx.query( `INSERT INTO atlas.product_image_suite_references (workspace_id, suite_id, asset_id, position) VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3)`, [row.id, assetId, position], ); await this.operation( tx, row.id, purpose === 'plan' ? 'image_suite_plan' : 'image_suite', userId, ); if (purpose === 'plan') await this.outbox( tx, 'image_suite.plan.requested', row.id, { suite_id: row.id }, ); else for (const [index, shot] of shots.entries()) { const supporting = input.reference_asset_ids.slice(1); // 默认方案轮换全部辅助底图,避免后面的参考角度始终未被使用。 const selected = shot.reference_asset_ids ?? [ input.reference_asset_ids[0], ...Array.from( { length: Math.min(2, supporting.length) }, (_, offset) => supporting[ (index + offset) % supporting.length ], ), ]; const settings = createMediaAiEditRequestSchema.parse({ product_id: productId, reference_asset_ids: selected.slice(1), mode: 'edit', ratio: input.ratio, target_language: input.language, background: shot.category === 'white' ? 'white' : 'auto', style_set: input.style_set, style_prompt: input.style_prompt, prompt: `${shot.theme}\n${shot.description}\n资料依据:${shot.evidence}`, suite_context: { type: shot.type, product_name: input.product_name, color: input.color, dimensions: input.dimensions, weight: input.weight, collected_info: input.collected_info, selling_points: input.selling_points, }, }); const id = randomUUID(); await tx.query( `INSERT INTO atlas.media_ai_edits (id, workspace_id, product_id, suite_id, suite_shot_id, suite_position, source_asset_id, reference_asset_ids, settings, created_by_user_id) VALUES ($1, current_setting('app.workspace_id')::uuid, $2, $3, $4, $5, $6, $7::uuid[], $8::jsonb, $9)`, [ id, productId, row.id, shot.id, index, selected[0], selected.slice(1), JSON.stringify(settings), userId, ], ); await this.operation(tx, id, 'media_ai_edit', userId); await this.outbox(tx, 'media.ai_edit.requested', id, { edit_id: id, }); } return { status: 202, body: { data: this.resource(row) } }; }, ); } public async read( workspaceId: string, productId: string, id: string, ): Promise { const { row, children } = await this.database.workspaceTransaction( workspaceId, async (tx) => { const row = await this.find(tx, productId, id); const children = await tx.query<{ id: string; suite_shot_id: string; }>( 'SELECT id, suite_shot_id FROM atlas.media_ai_edits WHERE suite_id = $1 ORDER BY suite_position', [id], ); return { row, children: children.rows }; }, ); return { ...this.resource(row), items: await Promise.all( children.map(async (child) => ({ ...(await this.edits.read(workspaceId, child.id)), shot_id: child.suite_shot_id, edit_id: child.id, })), ), }; } public async list( workspaceId: string, productId: string, query: ListProductImageTasksQuery, ) { let cursor: { at: string; id: string } | null = null; if (query.cursor) { try { cursor = JSON.parse( Buffer.from(query.cursor, 'base64url').toString(), ); if ( !cursor || typeof cursor.at !== 'string' || typeof cursor.id !== 'string' || !Number.isFinite(Date.parse(cursor.at)) || !/^[0-9a-f-]{36}$/i.test(cursor.id) ) throw new Error(); } catch { throw new ApiError(422, 'VALIDATION_FAILED', '分页游标无效'); } } const rows = await this.database.workspaceTransaction( workspaceId, async (tx) => { const found = await tx.query<{ id: string }>( 'SELECT id FROM atlas.products WHERE id = $1', [productId], ); if (!found.rows[0]) throw new ApiError(404, 'RESOURCE_NOT_FOUND', '商品不存在'); return ( await tx.query( `SELECT * FROM atlas.product_image_suites WHERE product_id = $1 AND ($2::timestamptz IS NULL OR (created_at, id) < ($2::timestamptz, $3::uuid)) ORDER BY created_at DESC, id DESC LIMIT $4`, [ productId, cursor?.at ?? null, cursor?.id ?? null, query.limit + 1, ], ) ).rows; }, ); const page = rows.slice(0, query.limit), last = page.at(-1), hasMore = rows.length > query.limit; return { data: page.map((row) => this.resource(row)), meta: { has_more: hasMore, next_cursor: hasMore && last ? Buffer.from( JSON.stringify({ at: iso(last.created_at), id: last.id, }), ).toString('base64url') : null, }, }; } public async action( workspaceId: string, userId: string, productId: string, id: string, key: string, action: 'cancel' | 'retry', ) { return this.idempotency.run( { workspaceId, userId, method: 'POST', path: `/api/products/${productId}/ai/image-suites/${id}/${action}`, key, request: {}, }, async (tx) => { // 与 Worker 一致先锁子任务再锁父任务,避免取消与结果回写形成相反锁序。 const children = await tx.query<{ id: string; status: string }>( 'SELECT id, status FROM atlas.media_ai_edits WHERE suite_id = $1 ORDER BY id FOR UPDATE', [id], ); const row = await this.find(tx, productId, id, true); if (action === 'cancel') { if (['queued', 'running'].includes(row.status)) { await tx.query( "UPDATE atlas.product_image_suites SET status = 'cancelled', completed_at = now() WHERE id = $1", [id], ); await tx.query( "UPDATE atlas.media_ai_edits SET status = 'cancelled', completed_at = now() WHERE suite_id = $1 AND status IN ('queued', 'running')", [id], ); await tx.query( "UPDATE atlas.operations SET status = 'cancelled', completed_at = now() WHERE (id = $1 OR id = ANY($2::uuid[])) AND status IN ('queued', 'running')", [id, children.rows.map((child) => child.id)], ); } } else { if ( row.purpose !== 'generate' || ['queued', 'running'].includes(row.status) ) throw new ApiError( 409, 'INVALID_TASK_STATE', '当前任务不能重试,请等待生成完成', ); const failed = children.rows.filter((child) => ['failed', 'cancelled'].includes(child.status), ); if (failed.length) { await tx.query( "UPDATE atlas.product_image_suites SET status = 'queued', error_code = NULL, completed_at = NULL WHERE id = $1", [id], ); await tx.query( "UPDATE atlas.operations SET status = 'queued', error_code = NULL, error_message = NULL, completed_at = NULL WHERE id = $1", [id], ); for (const child of failed) { const attempt = await tx.query<{ generation_attempt: number; }>( "UPDATE atlas.media_ai_edits SET status = 'queued', provider_task_id = NULL, error_code = NULL, completed_at = NULL, started_at = NULL, created_at = now(), generation_attempt = generation_attempt + 1 WHERE id = $1 RETURNING generation_attempt", [child.id], ); await tx.query( "UPDATE atlas.operations SET status = 'queued', result = '{}'::jsonb, error_code = NULL, error_message = NULL, completed_at = NULL WHERE id = $1", [child.id], ); await this.outbox( tx, 'media.ai_edit.requested', child.id, { edit_id: child.id, attempt: attempt.rows[0].generation_attempt, }, randomUUID(), ); } } } return { status: 200, body: { data: this.resource(await this.find(tx, productId, id)), }, }; }, ); } private async find( tx: DatabaseTransaction, productId: string, id: string, lock = false, ) { const result = await tx.query( `SELECT * FROM atlas.product_image_suites WHERE id = $1 AND product_id = $2 ${lock ? 'FOR UPDATE' : ''}`, [id, productId], ); if (!result.rows[0]) throw new ApiError(404, 'RESOURCE_NOT_FOUND', '套图任务不存在'); return result.rows[0]; } private resource(row: SuiteRow): ImageSuite { return { id: row.id, product_id: row.product_id, variant_id: row.variant_id, purpose: row.purpose, status: row.status, input: row.input, shots: row.shots as ImageSuiteShot[], items: [], error_code: row.error_code, created_at: iso(row.created_at), completed_at: row.completed_at ? iso(row.completed_at) : null, operation: { id: row.id, kind: row.purpose === 'plan' ? 'image_suite_plan' : 'image_suite', status: row.status === 'partial' ? 'failed' : row.status, created_at: iso(row.created_at), completed_at: row.completed_at ? iso(row.completed_at) : null, error: row.error_code ? { code: row.error_code, message: row.purpose === 'plan' ? (imageSuitePlanErrorMessages[ row.error_code ] ?? imageSuitePlanErrorMessages.AI_SUITE_PLAN_FAILED) : '部分图片未完成,可重试失败项', } : null, }, }; } private async operation( tx: DatabaseTransaction, id: string, kind: string, userId: string, ) { await tx.query( `INSERT INTO atlas.operations (id, workspace_id, kind, resource_type, resource_id, created_by_user_id) VALUES ($1, current_setting('app.workspace_id')::uuid, $2, $2, $1, $3)`, [id, kind, userId], ); } private async outbox( tx: DatabaseTransaction, event: string, id: string, payload: Record, dedup = id, ) { await tx.query( `INSERT INTO atlas.outbox_events (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key) VALUES (current_setting('app.workspace_id')::uuid, 'image_suite', $1, $2, $3::jsonb, $4)`, [id, event, JSON.stringify(payload), dedup], ); } }