import { randomUUID } from 'node:crypto'; import { createApplicationDatabase, type Database, type DatabaseTransaction, } from '@atlas/database'; import { createImageSuiteSchema, imageSuiteInputSchema, type ImageSuite, } from '@atlas/contracts'; import { describe, expect, it } from 'vitest'; import { IdempotencyService } from '../src/common/idempotency.service.js'; import { ImageSuiteService } from '../src/media/image-suite.service.js'; import { MediaAiEditService } from '../src/media/media-ai-edit.service.js'; import { ProductImageTaskService } from '../src/media/product-image-task.service.js'; /** 所有实体和 Outbox 都只存在于最终回滚的事务，验证不调用实际收费生图。 */ describe.skipIf(!process.env.DATABASE_URL)( '套图数据库隔离、幂等与状态汇总', () => { it('创建/部分失败/取消/重试保留成功图片，FK 与 RLS 拒绝越权引用', async () => { const db = createApplicationDatabase(process.env.DATABASE_URL!); const rollback = Symbol('rollback'); try { await db.transaction(async (tx) => { const workspaceId = randomUUID(), otherWorkspace = randomUUID(), userId = randomUUID(); await tx.query( "INSERT INTO atlas.users (id, email, password_hash) VALUES ($1, $2, 'test-only')", [userId, `${userId}@example.test`], ); for (const id of [workspaceId, otherWorkspace]) await tx.query( "INSERT INTO atlas.workspaces (id, name, slug) VALUES ($1::uuid, '套图测试', $1::text)", [id], ); const inWorkspace = async (id: string) => tx.query( "SELECT set_config('app.workspace_id', $1, true)", [id], ); await inWorkspace(workspaceId); const productId = randomUUID(), variantId = randomUUID(), itemId = randomUUID(), assetId = randomUUID(), outputId = randomUUID(); await tx.query( "INSERT INTO atlas.collection_items (id, workspace_id, source_kind, source_item_key) VALUES ($1::uuid, $2, 'manual', $1::text)", [itemId, workspaceId], ); await tx.query( 'INSERT INTO atlas.products (id, workspace_id, collection_item_id) VALUES ($1, $2, $3)', [productId, workspaceId, itemId], ); await tx.query( 'INSERT INTO atlas.product_variants (id, workspace_id, product_id) VALUES ($1, $2, $3)', [variantId, workspaceId, productId], ); const asset = async (id: string, ws: string) => tx.query( `INSERT INTO atlas.media_assets (id, workspace_id, storage_provider, bucket, object_key, content_type, byte_size, checksum_sha256, width_px, height_px) VALUES ($1::uuid, $2, 'local_minio', 'test', $1::text, 'image/jpeg', 100, $1::text, 100, 100)`, [id, ws], ); await asset(assetId, workspaceId); await asset(outputId, workspaceId); await inWorkspace(otherWorkspace); const foreignAsset = randomUUID(); await asset(foreignAsset, otherWorkspace); await inWorkspace(workspaceId); let queryQueue: Promise = Promise.resolve(); const scoped = { workspaceTransaction: ( ws: string, work: ( transaction: DatabaseTransaction, ) => Promise, ) => { const result = queryQueue.then(async () => { await inWorkspace(ws); return work(tx); }); queryQueue = result.then( () => undefined, () => undefined, ); return result; }, } as Database; const idempotency = new IdempotencyService(scoped, { IDEMPOTENCY_KEY_SECRET: 'suite-test-secret', } as never); const edits = new MediaAiEditService(scoped, idempotency, { accessUrl: async () => ({ url: 'https://example.test/preview', expires_at: '2099-01-01T00:00:00Z', }), } as never); const suites = new ImageSuiteService( scoped, idempotency, edits, ); const input = imageSuiteInputSchema.parse({ variant_id: variantId, reference_asset_ids: [assetId], product_name: '台灯', }); const shots = ['white', 'scene', 'selling'].map( (category, i) => ({ id: randomUUID(), category, type: ['white_background', 'lifestyle', 'detail'][ i ], theme: `方案${i}`, description: '依据真实底图展示商品', }), ); const request = createImageSuiteSchema.parse({ ...input, shots, }); const key = randomUUID(); const created = await suites.create( workspaceId, userId, productId, key, request, 'generate', ); const id = (created.body as { data: ImageSuite }).data.id; expect( ( await suites.create( workspaceId, userId, productId, key, request, 'generate', ) ).body, ).toEqual(created.body); await expect( suites.create( workspaceId, userId, productId, key, { ...request, product_name: '不同请求' }, 'generate', ), ).rejects.toMatchObject({ status: 409, response: { error: { code: 'IDEMPOTENCY_KEY_REUSED' } }, }); const children = ( await tx.query<{ id: string }>( 'SELECT id FROM atlas.media_ai_edits WHERE suite_id = $1 ORDER BY suite_position', [id], ) ).rows; expect(children).toHaveLength(3); expect( ( await tx.query( 'SELECT id FROM atlas.outbox_events WHERE aggregate_id = ANY($1::uuid[])', [children.map((child) => child.id)], ) ).rows, ).toHaveLength(3); await tx.query( "UPDATE atlas.media_ai_edits SET status = 'succeeded', output_asset_id = $2 WHERE id = $1", [children[0].id, outputId], ); await tx.query( "UPDATE atlas.media_ai_edits SET status = 'failed', error_code = 'TEST_FAILURE' WHERE id = $1", [children[1].id], ); expect( (await suites.read(workspaceId, productId, id)).status, ).toBe('running'); await suites.action( workspaceId, userId, productId, id, randomUUID(), 'cancel', ); expect( (await suites.read(workspaceId, productId, id)).status, ).toBe('cancelled'); await suites.action( workspaceId, userId, productId, id, randomUUID(), 'retry', ); const retried = await suites.read( workspaceId, productId, id, ); expect(retried.items.map((item) => item.status)).toEqual([ 'succeeded', 'queued', 'queued', ]); expect(retried.items[0].output_asset?.id).toBe(outputId); const attempts = ( await tx.query<{ generation_attempt: number }>( 'SELECT generation_attempt FROM atlas.media_ai_edits WHERE suite_id = $1 ORDER BY suite_position', [id], ) ).rows; expect( attempts.map((item) => item.generation_attempt), ).toEqual([1, 2, 2]); await tx.query( "UPDATE atlas.media_ai_edits SET status = 'failed', error_code = 'TEST_FAILURE' WHERE id = ANY($1::uuid[])", [children.slice(1).map((child) => child.id)], ); expect( (await suites.read(workspaceId, productId, id)).status, ).toBe('partial'); const tasks = await new ProductImageTaskService( scoped, ).list(workspaceId, productId, { limit: 20 }); expect(tasks.data).toMatchObject([ { id, kind: 'ai_suite', status: 'partial', total: 3, completed: 1, output_asset_ids: [outputId], }, ]); expect(tasks.data).toHaveLength(1); await expect( suites.read(otherWorkspace, productId, id), ).rejects.toMatchObject({ status: 404 }); await expect( suites.create( workspaceId, userId, productId, randomUUID(), { ...input, reference_asset_ids: [foreignAsset] }, 'plan', ), ).rejects.toMatchObject({ status: 404 }); const forbidden = async ( sql: string, params: unknown[], code: string, ) => { await tx.query('SAVEPOINT forbidden_write'); try { await tx.query(sql, params); throw new Error('本应拒绝的写入成功'); } catch (error) { expect((error as { code: string }).code).toBe(code); } finally { await tx.query( 'ROLLBACK TO SAVEPOINT forbidden_write', ); } }; await forbidden( 'UPDATE atlas.product_image_suites SET input = input || \'{"color":"篡改"}\'::jsonb WHERE id = $1', [id], '23514', ); await forbidden( 'UPDATE atlas.media_ai_edits SET suite_position = NULL WHERE id = $1', [children[0].id], '23514', ); await forbidden( 'UPDATE atlas.media_ai_edits SET source_asset_id = $2 WHERE id = $1', [children[0].id, foreignAsset], '23514', ); const foreignReferenceParent = await suites.create( workspaceId, userId, productId, randomUUID(), input, 'plan', ); const foreignParentId = ( foreignReferenceParent.body as { data: ImageSuite } ).data.id; await forbidden( 'UPDATE atlas.product_image_suite_references SET asset_id = $2 WHERE suite_id = $1', [foreignParentId, foreignAsset], '23514', ); await inWorkspace(otherWorkspace); expect( ( await tx.query( 'SELECT id FROM atlas.product_image_suites WHERE id = $1', [id], ) ).rows, ).toHaveLength(0); throw rollback; }); } catch (error) { if (error !== rollback) throw error; } finally { await db.close(); } }); }, );