Update seller store experience

2026-09-08 16:19:38 +08:00
parent ceb644a3f9
commit 319887726e
25 changed files with 31417 additions and 127 deletions
@@ -1,6 +1,6 @@
 import { Body, Controller, Get, Headers, Inject, Param, Patch, Post, Query, Req, Res } from '@nestjs/common';
 import { z } from 'zod';
-import { bulkDeleteCollectionItemsRequestSchema, clearCollectionItemsRequestSchema, createCollectionItemRequestSchema, listCollectionItemsQuerySchema, retryCollectionAssetsRequestSchema, updateCollectionItemRequestSchema, type Workspace } from '@atlas/contracts';
+import { bulkDeleteCollectionItemsRequestSchema, clearCollectionItemsRequestSchema, createCollectionItemRequestSchema, listCollectionItemsQuerySchema, materializeCollectionAssetsRequestSchema, retryCollectionAssetsRequestSchema, updateCollectionItemRequestSchema, type Workspace } from '@atlas/contracts';
 import { ApiError } from '../common/api-error.js';
 import type { AuthenticatedRequest } from '../auth/auth.guard.js';
 import { CurrentWorkspace } from '../workspace/current-workspace.decorator.js';
@@ -32,6 +32,9 @@ export class CollectionController {
  @Post(':id/asset-retries') async retry(@Param('id') id: string, @Body() body: unknown, @Headers('idempotency-key') key: string | undefined, @Req() request: AuthenticatedRequest, @CurrentWorkspace() workspace: Workspace, @Res({ passthrough: true }) response: Response) {
    const result = await this.collection.retryAssets(workspace.id, request.auth.userId, requireUuid('collection_item_id', id), requireKey(key), parse(retryCollectionAssetsRequestSchema, body)); response.status(result.status); return result.body;
  }
+  @Post(':id/asset-materializations') async materialize(@Param('id') id: string, @Body() body: unknown, @Headers('idempotency-key') key: string | undefined, @Req() request: AuthenticatedRequest, @CurrentWorkspace() workspace: Workspace, @Res({ passthrough: true }) response: Response) {
+    const result = await this.collection.materializeAssets(workspace.id, request.auth.userId, requireUuid('collection_item_id', id), requireKey(key), parse(materializeCollectionAssetsRequestSchema, body)); response.status(result.status); return result.body;
+  }
 }
 function parse<T extends { safeParse: (input: unknown) => { success: boolean; data?: unknown; error?: { issues: Array<{ path: PropertyKey[]; message: string }> } } }>(schema: T, input: unknown) { const result = schema.safeParse(input); if (result.success) return result.data as never; throw new ApiError(422, 'VALIDATION_FAILED', '请求参数不符合要求', result.error?.issues.map((x) => ({ field: x.path.join('.'), code: 'INVALID_FIELD', message: x.message }))); }
 function requireUuid(field: string, value: string) { if (z.uuid().safeParse(value).success) return value; throw new ApiError(422, 'VALIDATION_FAILED', '请求参数不符合要求', [{ field, code: 'INVALID_FIELD', message: `${field} 必须是 UUID` }]); }
@@ -1,6 +1,6 @@
 import { randomUUID } from 'node:crypto';
 import { Inject, Injectable } from '@nestjs/common';
-import type { CreateCollectionItemRequest, ListCollectionItemsQuery, UpdateCollectionItemRequest } from '@atlas/contracts';
+import type { CreateCollectionItemRequest, ListCollectionItemsQuery, MaterializeCollectionAssetsRequest, UpdateCollectionItemRequest } from '@atlas/contracts';
 import type { Database, DatabaseTransaction } from '@atlas/database';
 import { ApiError } from '../common/api-error.js';
 import { IdempotencyService } from '../common/idempotency.service.js';
@@ -37,7 +37,7 @@ export class CollectionService {
      const item = await this.requireItem(tx, id);
      const source = await tx.query<{ normalized_payload: unknown; raw_evidence: unknown; parser_version: string | null; collected_at: Date | string | null; source_snapshot_revision: string | number }>('SELECT normalized_payload, raw_evidence, parser_version, collected_at, source_snapshot_revision FROM atlas.collection_items WHERE id = $1 AND deleted_at IS NULL', [id]);
      const variants = await tx.query<{ id: string; source_variant_key: string | null; source_sku: string | null; display_name: string | null; sort_order: number }>('SELECT id, source_variant_key, source_sku, display_name, sort_order FROM atlas.product_variants WHERE product_id = $1 AND deleted_at IS NULL ORDER BY sort_order, id', [item.product_id]);
-      const transfers = await tx.query<{ id: string; source_variant_id: string | null; role: string; source_url: string; status: string; asset_id: string | null; error_payload: unknown; created_at: Date | string; completed_at: Date | string | null }>('SELECT id, source_variant_id, role::text AS role, source_url, status::text AS status, asset_id, error_payload, created_at, completed_at FROM atlas.collection_asset_transfers WHERE collection_item_id = $1 AND source_snapshot_revision = $2 ORDER BY created_at, id', [id, source.rows[0]?.source_snapshot_revision ?? 1]);
+      const transfers = await tx.query<{ id: string; source_variant_id: string | null; media_type: 'image' | 'video'; poster_url: string | null; role: string; source_url: string; status: string; asset_id: string | null; error_payload: unknown; created_at: Date | string; completed_at: Date | string | null }>('SELECT id, source_variant_id, media_type::text AS media_type, poster_url, role::text AS role, source_url, status::text AS status, asset_id, error_payload, created_at, completed_at FROM atlas.collection_asset_transfers WHERE collection_item_id = $1 AND source_snapshot_revision = $2 ORDER BY created_at, id', [id, source.rows[0]?.source_snapshot_revision ?? 1]);
      return { ...item, normalized_payload: source.rows[0]?.normalized_payload ?? {}, raw_evidence: source.rows[0]?.raw_evidence ?? {}, parser_version: source.rows[0]?.parser_version ?? null, collected_at: source.rows[0]?.collected_at ? iso(source.rows[0].collected_at) : null, variants: variants.rows, asset_transfers: transfers.rows.map((transfer) => ({ ...transfer, created_at: iso(transfer.created_at), completed_at: transfer.completed_at ? iso(transfer.completed_at) : null })) };
    });
  }
@@ -51,6 +51,7 @@ export class CollectionService {
  async bulkDelete(workspaceId: string, userId: string, key: string, ids: { ids: string[] }) { return this.removeMany(workspaceId, userId, key, '/api/collection-items/bulk-delete', ids, false); }
  async clear(workspaceId: string, userId: string, key: string, input: { confirm: true; expected_active_count: number }) { return this.removeMany(workspaceId, userId, key, '/api/collection-items/clear', input, true); }
  async retryAssets(workspaceId: string, userId: string, id: string, key: string, input: { transfer_ids: string[] }) { return this.idempotency.run({ workspaceId, userId, method: 'POST', path: `/api/collection-items/${id}/asset-retries`, key, request: input }, async (tx) => { await this.requireItem(tx, id); const result = await tx.query<{ source_snapshot_revision: string | number }>(`UPDATE atlas.collection_asset_transfers t SET status = 'queued', error_payload = NULL, updated_at = now() FROM atlas.collection_items c WHERE t.collection_item_id = c.id AND t.collection_item_id = $1 AND t.id = ANY($2::uuid[]) AND t.source_snapshot_revision = c.source_snapshot_revision AND t.status IN ('failed', 'cancelled') RETURNING c.source_snapshot_revision`, [id, input.transfer_ids]); if (result.rowCount !== input.transfer_ids.length) throw new ApiError(422, 'ASSET_RETRY_INVALID', '只能重试当前采集快照中失败或已取消的素材任务'); const operation = await this.operation(tx, userId, 'collection_asset_retry', id); await this.event(tx, id, 'collection.assets.retry.requested', { operation_id: operation.id, collection_item_id: id, source_snapshot_revision: Number(result.rows[0].source_snapshot_revision) }); return { status: 202, body: { data: { operation } } }; }); }
+  async materializeAssets(workspaceId: string, userId: string, id: string, key: string, input: MaterializeCollectionAssetsRequest) { return this.idempotency.run({ workspaceId, userId, method: 'POST', path: `/api/collection-items/${id}/asset-materializations`, key, request: input }, async (tx) => { await this.requireItem(tx, id); const result = await tx.query<{ source_snapshot_revision: string | number }>(`UPDATE atlas.collection_asset_transfers t SET status = 'queued', error_payload = NULL, completed_at = NULL, updated_at = now() FROM atlas.collection_items c WHERE t.collection_item_id = c.id AND t.collection_item_id = $1 AND t.id = ANY($2::uuid[]) AND t.source_snapshot_revision = c.source_snapshot_revision AND t.status = 'pending_selection' RETURNING c.source_snapshot_revision`, [id, input.transfer_ids]); if (result.rowCount !== input.transfer_ids.length) throw new ApiError(422, 'ASSET_MATERIALIZATION_INVALID', '只能转存当前采集快照中尚未转存的来源媒体'); const operation = await this.operation(tx, userId, 'collection_asset_materialization', id); await this.event(tx, operation.id, 'collection.assets.materialize.requested', { operation_id: operation.id, collection_item_id: id, source_snapshot_revision: Number(result.rows[0].source_snapshot_revision) }); return { status: 202, body: { data: { operation } } }; }); }
  private async removeMany(workspaceId: string, userId: string, key: string, path: string, input: unknown, clear: boolean) { return this.idempotency.run({ workspaceId, userId, method: 'POST', path, key, request: input }, async (tx) => { const requested = input as { ids?: string[]; expected_active_count?: number }; const count = await tx.query<{ count: string }>('SELECT count(*)::text AS count FROM atlas.collection_items WHERE deleted_at IS NULL'); if (clear && Number(count.rows[0].count) !== requested.expected_active_count) throw new ApiError(409, 'COLLECTION_COUNT_CHANGED', '采集箱数量已变化，请重新确认'); const deleted = await tx.query<{ id: string }>(clear ? `UPDATE atlas.collection_items SET status = 'deleted', deleted_at = now(), revision = revision + 1, updated_at = now() WHERE deleted_at IS NULL RETURNING id` : `UPDATE atlas.collection_items SET status = 'deleted', deleted_at = now(), revision = revision + 1, updated_at = now() WHERE id = ANY($1::uuid[]) AND deleted_at IS NULL RETURNING id`, clear ? [] : [requested.ids]); if (!clear && deleted.rows.length !== requested.ids?.length) throw new ApiError(404, 'RESOURCE_NOT_FOUND', '部分采集记录不存在'); const operation = await this.operation(tx, userId, 'collection_cleanup', deleted.rows[0]?.id ?? null); await this.event(tx, operation.id, 'collection.cleanup.requested', { operation_id: operation.id, collection_item_ids: deleted.rows.map((row) => row.id) }); return { status: 202, body: { data: { deleted_count: deleted.rows.length, operation } } }; }); }
  private async operation(tx: DatabaseTransaction, userId: string, kind: string, resourceId: string | null) { const row = await tx.query<{ id: string; status: string; created_at: Date | string }>(`INSERT INTO atlas.operations (workspace_id, kind, resource_type, resource_id, created_by_user_id) VALUES (current_setting('app.workspace_id')::uuid, $1, 'collection_item', $2, $3) RETURNING id, status, created_at`, [kind, resourceId, userId]); return { id: row.rows[0].id, kind, status: row.rows[0].status, created_at: iso(row.rows[0].created_at), completed_at: null, error: null }; }
  private async event(tx: DatabaseTransaction, aggregateId: string, type: string, payload: unknown) { 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, 'collection_item', $1, $2, $3, $4)`, [aggregateId, type, payload, `${type}:${aggregateId}`]); }
@@ -9,39 +9,110 @@ import { DATABASE } from '../database/database.module.js';
 export class ExtensionService {
    public constructor(
        @Inject(DATABASE) private readonly database: Database,
-        @Inject(IdempotencyService) private readonly idempotency: IdempotencyService,
+        @Inject(IdempotencyService)
+        private readonly idempotency: IdempotencyService,
    ) {}

-    public async ingest(workspaceId: string, userId: string, key: string, payload: CollectedProductPayload) {
-        return this.idempotency.run({ workspaceId, userId, method: 'POST', path: '/api/extension/materials', key, request: payload }, async (tx) => {
-            const platform = await tx.query<{ id: string; code: string }>(`SELECT id, code FROM atlas.platforms
-              WHERE code = $1 AND status = 'active' AND coalesce((capabilities->>'can_be_source')::boolean, false)`, [payload.source.platform.toUpperCase()]);
-            if (!platform.rows[0]) throw new ApiError(422, 'SOURCE_PLATFORM_UNAVAILABLE', '该来源平台当前不可采集');
-            const existing = await this.findExistingSourceItem(
-                tx,
+    public async ingest(
+        workspaceId: string,
+        userId: string,
+        key: string,
+        payload: CollectedProductPayload,
+    ) {
+        return this.idempotency.run(
+            {
                workspaceId,
-                platform.rows[0].id,
-                payload,
-            );
-            if (existing) {
-                await this.replaceSourceSnapshot(
+                userId,
+                method: 'POST',
+                path: '/api/extension/materials',
+                key,
+                request: payload,
+            },
+            async (tx) => {
+                const platform = await tx.query<{ id: string; code: string }>(
+                    `SELECT id, code FROM atlas.platforms
+              WHERE code = $1 AND status = 'active' AND coalesce((capabilities->>'can_be_source')::boolean, false)`,
+                    [payload.source.platform.toUpperCase()],
+                );
+                if (!platform.rows[0])
+                    throw new ApiError(
+                        422,
+                        'SOURCE_PLATFORM_UNAVAILABLE',
+                        '该来源平台当前不可采集',
+                    );
+                const existing = await this.findExistingSourceItem(
                    tx,
-                    existing.id,
-                    existing.product_id,
+                    workspaceId,
+                    platform.rows[0].id,
                    payload,
                );
-                return { status: 200, body: { data: await this.materialResponse(tx, existing.id, existing.product_id, true, true) } };
-            }
-            const { status, sourcePrice } = sourceState(payload);
-            const item = await tx.query<{ id: string }>(`INSERT INTO atlas.collection_items
+                if (existing) {
+                    await this.replaceSourceSnapshot(
+                        tx,
+                        workspaceId,
+                        existing.id,
+                        existing.product_id,
+                        payload,
+                    );
+                    return {
+                        status: 200,
+                        body: {
+                            data: await this.materialResponse(
+                                tx,
+                                existing.id,
+                                existing.product_id,
+                                true,
+                                true,
+                            ),
+                        },
+                    };
+                }
+                const { status, sourcePrice } = sourceState(payload);
+                const item = await tx.query<{ id: string }>(
+                    `INSERT INTO atlas.collection_items
                (workspace_id, source_kind, source_platform_id, source_item_key, source_url, source_title, source_currency_code, source_min_price, source_max_price, status, raw_evidence, normalized_payload, parser_version, collected_at)
                VALUES (current_setting('app.workspace_id')::uuid, 'platform', $1, $2, $3, $4, $5, $6, $6, $7::atlas.collection_item_status, $8, $9, $10, $11) RETURNING id`,
-                [platform.rows[0].id, payload.source.itemKey, payload.source.canonicalUrl, payload.common.title?.value ?? null, sourcePrice?.currencyCode ?? null, sourcePrice?.amount ?? null, status, JSON.stringify(payload.evidence), JSON.stringify(payload), payload.source.parserVersion, payload.source.collectedAt]);
-            const product = await tx.query<{ id: string }>(`INSERT INTO atlas.products (workspace_id, collection_item_id, source_title)
-                VALUES (current_setting('app.workspace_id')::uuid, $1, $2) RETURNING id`, [item.rows[0].id, payload.common.title?.value ?? null]);
-            await this.writeSourceSnapshotRows(tx, item.rows[0].id, product.rows[0].id, 1, payload);
-            return { status: 201, body: { data: await this.materialResponse(tx, item.rows[0].id, product.rows[0].id, false, false) } };
-        });
+                    [
+                        platform.rows[0].id,
+                        payload.source.itemKey,
+                        payload.source.canonicalUrl,
+                        payload.common.title?.value ?? null,
+                        sourcePrice?.currencyCode ?? null,
+                        sourcePrice?.amount ?? null,
+                        status,
+                        JSON.stringify(payload.evidence),
+                        JSON.stringify(payload),
+                        payload.source.parserVersion,
+                        payload.source.collectedAt,
+                    ],
+                );
+                const product = await tx.query<{ id: string }>(
+                    `INSERT INTO atlas.products (workspace_id, collection_item_id, source_title)
+                VALUES (current_setting('app.workspace_id')::uuid, $1, $2) RETURNING id`,
+                    [item.rows[0].id, payload.common.title?.value ?? null],
+                );
+                await this.writeSourceSnapshotRows(
+                    tx,
+                    workspaceId,
+                    item.rows[0].id,
+                    product.rows[0].id,
+                    1,
+                    payload,
+                );
+                return {
+                    status: 201,
+                    body: {
+                        data: await this.materialResponse(
+                            tx,
+                            item.rows[0].id,
+                            product.rows[0].id,
+                            false,
+                            false,
+                        ),
+                    },
+                };
+            },
+        );
    }

    /**
@@ -70,7 +141,10 @@ export class ExtensionService {
        const sourceSkus = [
            ...new Set(
                payload.variants
-                    .map((variant) => variant.sourceSku ?? variant.sourceVariantKey)
+                    .map(
+                        (variant) =>
+                            variant.sourceSku ?? variant.sourceVariantKey,
+                    )
                    .filter((sku) => sku.trim().length > 0),
            ),
        ];
@@ -109,55 +183,201 @@ export class ExtensionService {
        return byVariantOverlap.rows[0] ?? null;
    }

-    private async replaceSourceSnapshot(tx: { query: Database['query'] }, collectionItemId: string, productId: string, payload: CollectedProductPayload) {
+    private async replaceSourceSnapshot(
+        tx: { query: Database['query'] },
+        workspaceId: string,
+        collectionItemId: string,
+        productId: string,
+        payload: CollectedProductPayload,
+    ) {
        const { status, sourcePrice } = sourceState(payload);
-        // 不删除旧任务记录：运行中的 Worker 可能已拿到它。取消并保留审计记录，
+        // 不删除旧候选/任务记录：运行中的 Worker 可能已拿到它。取消并保留审计记录，
        // 新快照在查询和 Worker 中均按 source_snapshot_revision 隔离。
-        await tx.query(`UPDATE atlas.collection_asset_transfers SET status = 'cancelled', completed_at = now(), updated_at = now()
-          WHERE collection_item_id = $1 AND status IN ('queued', 'running')`, [collectionItemId]);
-        const item = await tx.query<{ source_snapshot_revision: string | number }>(`UPDATE atlas.collection_items
+        await tx.query(
+            `UPDATE atlas.collection_asset_transfers SET status = 'cancelled', completed_at = now(), updated_at = now()
+          WHERE collection_item_id = $1 AND status IN ('pending_selection', 'queued', 'running')`,
+            [collectionItemId],
+        );
+        const item = await tx.query<{
+            source_snapshot_revision: string | number;
+        }>(
+            `UPDATE atlas.collection_items
          SET source_url = $1, source_title = $2, source_currency_code = $3, source_min_price = $4, source_max_price = $4,
              status = $5::atlas.collection_item_status, raw_evidence = $6, normalized_payload = $7, parser_version = $8,
              collected_at = $9, source_snapshot_revision = source_snapshot_revision + 1, revision = revision + 1, updated_at = now()
          WHERE id = $10 AND deleted_at IS NULL RETURNING source_snapshot_revision`,
-            [payload.source.canonicalUrl, payload.common.title?.value ?? null, sourcePrice?.currencyCode ?? null, sourcePrice?.amount ?? null, status, JSON.stringify(payload.evidence), JSON.stringify(payload), payload.source.parserVersion, payload.source.collectedAt, collectionItemId]);
+            [
+                payload.source.canonicalUrl,
+                payload.common.title?.value ?? null,
+                sourcePrice?.currencyCode ?? null,
+                sourcePrice?.amount ?? null,
+                status,
+                JSON.stringify(payload.evidence),
+                JSON.stringify(payload),
+                payload.source.parserVersion,
+                payload.source.collectedAt,
+                collectionItemId,
+            ],
+        );
        const snapshotRevision = Number(item.rows[0]?.source_snapshot_revision);
-        if (!Number.isSafeInteger(snapshotRevision) || snapshotRevision < 1) throw new Error('采集来源快照版本更新失败');
-        await tx.query(`UPDATE atlas.products SET source_title = $1, updated_at = now(), version = version + 1 WHERE id = $2`, [payload.common.title?.value ?? null, productId]);
-        await tx.query('DELETE FROM atlas.product_variants WHERE product_id = $1', [productId]);
-        await this.writeSourceSnapshotRows(tx, collectionItemId, productId, snapshotRevision, payload);
+        if (!Number.isSafeInteger(snapshotRevision) || snapshotRevision < 1)
+            throw new Error('采集来源快照版本更新失败');
+        await tx.query(
+            `UPDATE atlas.products SET source_title = $1, updated_at = now(), version = version + 1 WHERE id = $2`,
+            [payload.common.title?.value ?? null, productId],
+        );
+        // The historical task rows must remain auditable after replacing the
+        // source snapshot. Clear only their optional variant pointer before
+        // deleting old variants. The original composite FK used ON DELETE SET
+        // NULL and PostgreSQL tried to null both key columns, including the
+        // required workspace_id.
+        await tx.query(
+            `UPDATE atlas.collection_asset_transfers
+             SET source_variant_id = NULL, updated_at = now()
+             WHERE product_id = $1 AND source_variant_id IS NOT NULL`,
+            [productId],
+        );
+        await tx.query(
+            'DELETE FROM atlas.product_variants WHERE product_id = $1',
+            [productId],
+        );
+        await this.writeSourceSnapshotRows(
+            tx,
+            workspaceId,
+            collectionItemId,
+            productId,
+            snapshotRevision,
+            payload,
+        );
        return snapshotRevision;
    }

-    private async writeSourceSnapshotRows(tx: { query: Database['query'] }, collectionItemId: string, productId: string, snapshotRevision: number, payload: CollectedProductPayload) {
-        const verifiedVariants = payload.variants.filter((variant) => variant.sourceVariantKey.trim().length > 0);
-        const variantRows = verifiedVariants.length ? verifiedVariants : [{ sourceVariantKey: null, sourceSku: undefined, displayName: '默认 SKU' }];
+    private async writeSourceSnapshotRows(
+        tx: { query: Database['query'] },
+        workspaceId: string,
+        collectionItemId: string,
+        productId: string,
+        snapshotRevision: number,
+        payload: CollectedProductPayload,
+    ) {
+        const verifiedVariants = payload.variants.filter(
+            (variant) => variant.sourceVariantKey.trim().length > 0,
+        );
+        const variantRows = verifiedVariants.length
+            ? verifiedVariants
+            : [
+                  {
+                      sourceVariantKey: null,
+                      sourceSku: undefined,
+                      displayName: '默认 SKU',
+                  },
+              ];
        const variants = new Map<string, string>();
        for (let index = 0; index < variantRows.length; index += 1) {
            const variant = variantRows[index];
-            const inserted = await tx.query<{ id: string }>(`INSERT INTO atlas.product_variants (workspace_id, product_id, source_variant_key, source_sku, display_name, sort_order)
-              VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4, $5) RETURNING id`, [productId, variant.sourceVariantKey, variant.sourceSku ?? null, variant.displayName, index]);
-            if (variant.sourceVariantKey) variants.set(variant.sourceVariantKey, inserted.rows[0].id);
+            const inserted = await tx.query<{ id: string }>(
+                `INSERT INTO atlas.product_variants (workspace_id, product_id, source_variant_key, source_sku, display_name, sort_order)
+              VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
+                [
+                    workspaceId,
+                    productId,
+                    variant.sourceVariantKey,
+                    variant.sourceSku ?? null,
+                    variant.displayName,
+                    index,
+                ],
+            );
+            if (variant.sourceVariantKey)
+                variants.set(variant.sourceVariantKey, inserted.rows[0].id);
        }
        for (const asset of payload.assets) {
-            await tx.query(`INSERT INTO atlas.collection_asset_transfers (workspace_id, collection_item_id, product_id, source_variant_id, role, source_url, status, error_payload, source_snapshot_revision)
-                VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4::atlas.media_candidate_role, $5, 'queued', $6, $7)`,
-                [collectionItemId, productId, asset.variantSourceKey ? variants.get(asset.variantSourceKey) ?? null : null, asset.role, asset.url, JSON.stringify({ relation: asset.relation, referer_origin: asset.refererOrigin ?? null, evidence_ref: asset.evidenceRef ?? null, sort_order: asset.sortOrder }), snapshotRevision]);
+            await tx.query(
+                `INSERT INTO atlas.collection_asset_transfers (workspace_id, collection_item_id, product_id, source_variant_id, media_type, poster_url, role, source_url, status, error_payload, source_snapshot_revision)
+                VALUES ($1, $2, $3, $4, $5::atlas.collection_media_type, $6, $7::atlas.media_candidate_role, $8, 'pending_selection', $9, $10)`,
+                [
+                    workspaceId,
+                    collectionItemId,
+                    productId,
+                    asset.variantSourceKey
+                        ? (variants.get(asset.variantSourceKey) ?? null)
+                        : null,
+                    asset.mediaType,
+                    asset.posterUrl ?? null,
+                    asset.role,
+                    asset.url,
+                    JSON.stringify({
+                        relation: asset.relation,
+                        referer_origin: asset.refererOrigin ?? null,
+                        evidence_ref: asset.evidenceRef ?? null,
+                        sort_order: asset.sortOrder,
+                    }),
+                    snapshotRevision,
+                ],
+            );
        }
-        if (payload.assets.length) 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, 'collection_item', $1, 'collection.assets.materialize.requested', jsonb_build_object('collection_item_id', $1::uuid, 'source_snapshot_revision', $2::bigint), $3)`, [collectionItemId, snapshotRevision, `collection.assets.materialize:${collectionItemId}:${snapshotRevision}`]);
-        else await tx.query(`UPDATE atlas.collection_items SET status = CASE WHEN status = 'needs_review' THEN status ELSE 'ready_for_edit' END, updated_at = now() WHERE id = $1 AND source_snapshot_revision = $2`, [collectionItemId, snapshotRevision]);
+        // Collecting records source-media candidates only. A later explicit
+        // selection queues the media and creates the materialisation event.
+        await tx.query(
+            `UPDATE atlas.collection_items SET status = CASE WHEN status = 'needs_review' THEN status ELSE 'ready_for_edit' END, updated_at = now() WHERE id = $1 AND source_snapshot_revision = $2`,
+            [collectionItemId, snapshotRevision],
+        );
    }

-    private async materialResponse(tx: { query: Database['query'] }, collectionItemId: string, productId: string, reused: boolean, refreshed: boolean) {
-        const count = await tx.query<{ total: string; queued: string; succeeded: string; failed: string }>(`SELECT count(*)::text AS total, count(*) FILTER (WHERE t.status = 'queued')::text AS queued, count(*) FILTER (WHERE t.status = 'succeeded')::text AS succeeded, count(*) FILTER (WHERE t.status = 'failed')::text AS failed FROM atlas.collection_asset_transfers t JOIN atlas.collection_items c ON c.id = t.collection_item_id WHERE t.collection_item_id = $1 AND t.source_snapshot_revision = c.source_snapshot_revision`, [collectionItemId]);
-        const status = await tx.query<{ status: string; source_title: string | null }>('SELECT status, source_title FROM atlas.collection_items WHERE id = $1', [collectionItemId]);
-        return { collection_item: { id: collectionItemId, status: status.rows[0]?.status ?? 'received', source_title: status.rows[0]?.source_title ?? null, variant_count: 0 }, product_id: productId, reused, refreshed, editor_entry_path: `/collection-test/${collectionItemId}`, materialization: { total: Number(count.rows[0]?.total ?? 0), queued: Number(count.rows[0]?.queued ?? 0), succeeded: Number(count.rows[0]?.succeeded ?? 0), failed: Number(count.rows[0]?.failed ?? 0) } };
+    private async materialResponse(
+        tx: { query: Database['query'] },
+        collectionItemId: string,
+        productId: string,
+        reused: boolean,
+        refreshed: boolean,
+    ) {
+        const count = await tx.query<{
+            total: string;
+            pending_selection: string;
+            queued: string;
+            succeeded: string;
+            failed: string;
+        }>(
+            `SELECT count(*)::text AS total, count(*) FILTER (WHERE t.status = 'pending_selection')::text AS pending_selection, count(*) FILTER (WHERE t.status = 'queued')::text AS queued, count(*) FILTER (WHERE t.status = 'succeeded')::text AS succeeded, count(*) FILTER (WHERE t.status = 'failed')::text AS failed FROM atlas.collection_asset_transfers t JOIN atlas.collection_items c ON c.id = t.collection_item_id WHERE t.collection_item_id = $1 AND t.source_snapshot_revision = c.source_snapshot_revision`,
+            [collectionItemId],
+        );
+        const status = await tx.query<{
+            status: string;
+            source_title: string | null;
+        }>(
+            'SELECT status, source_title FROM atlas.collection_items WHERE id = $1',
+            [collectionItemId],
+        );
+        return {
+            collection_item: {
+                id: collectionItemId,
+                status: status.rows[0]?.status ?? 'received',
+                source_title: status.rows[0]?.source_title ?? null,
+                variant_count: 0,
+            },
+            product_id: productId,
+            reused,
+            refreshed,
+            editor_entry_path: `/collection-test/${collectionItemId}`,
+            materialization: {
+                total: Number(count.rows[0]?.total ?? 0),
+                pending_selection: Number(count.rows[0]?.pending_selection ?? 0),
+                queued: Number(count.rows[0]?.queued ?? 0),
+                succeeded: Number(count.rows[0]?.succeeded ?? 0),
+                failed: Number(count.rows[0]?.failed ?? 0),
+            },
+        };
    }
-
 }

 function sourceState(payload: CollectedProductPayload) {
-    const verifiedVariants = payload.variants.filter((variant) => variant.sourceVariantKey.trim().length > 0);
-    return { status: verifiedVariants.length || !payload.variantAxes.length ? 'materializing' : 'needs_review', sourcePrice: payload.common.sourcePrice };
+    const verifiedVariants = payload.variants.filter(
+        (variant) => variant.sourceVariantKey.trim().length > 0,
+    );
+    return {
+        status:
+            verifiedVariants.length || !payload.variantAxes.length
+                ? 'ready_for_edit'
+                : 'needs_review',
+        sourcePrice: payload.common.sourcePrice,
+    };
 }
@@ -7,30 +7,326 @@ import { afterAll, beforeAll, describe, it } from 'vitest';
 import { DATABASE } from '../src/database/database.module.js';
 import { createApplication } from '../src/main.js';

-let app: NestExpressApplication; let database: Database; let baseUrl = ''; let workspaceId = ''; let userId = '';
-async function request(path: string, init: RequestInit = {}) { const response = await fetch(`${baseUrl}${path}`, { redirect: 'manual', ...init, headers: { 'Content-Type': 'application/json', ...init.headers } }); return { response, body: response.status === 204 || response.status === 302 ? undefined : await response.json() as any }; }
+let app: NestExpressApplication;
+let database: Database;
+let baseUrl = '';
+let workspaceId = '';
+let userId = '';
+async function request(path: string, init: RequestInit = {}) {
+    const response = await fetch(`${baseUrl}${path}`, {
+        redirect: 'manual',
+        ...init,
+        headers: { 'Content-Type': 'application/json', ...init.headers },
+    });
+    return {
+        response,
+        body:
+            response.status === 204 || response.status === 302
+                ? undefined
+                : ((await response.json()) as any),
+    };
+}

 describe('扩展授权与采集 HTTP 集成测试', () => {
-    beforeAll(async () => { Object.assign(process.env, { NODE_ENV: 'test', SWAGGER_ENABLED: 'false', AUTH_REGISTER_IP_LIMIT: '100', AUTH_REGISTER_EMAIL_LIMIT: '100' }); const created = await createApplication(); app = created.app; database = app.get<Database>(DATABASE); await app.listen(0, '127.0.0.1'); baseUrl = `http://127.0.0.1:${(app.getHttpServer().address() as AddressInfo).port}`; });
-    afterAll(async () => { await database.workspaceTransaction(workspaceId, async (tx) => { await tx.query('DELETE FROM atlas.idempotency_requests WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.outbox_events WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.collection_asset_transfers WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.product_variants WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.products WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.collection_items WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.extension_authorization_codes WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.extension_sessions WHERE workspace_id = $1', [workspaceId]); }); await database.query('DELETE FROM atlas.workspace_memberships WHERE user_id = $1', [userId]); await database.query('DELETE FROM atlas.workspaces WHERE id = $1', [workspaceId]); await database.query('DELETE FROM atlas.users WHERE id = $1', [userId]); await app.close(); });
+    beforeAll(async () => {
+        Object.assign(process.env, {
+            NODE_ENV: 'test',
+            SWAGGER_ENABLED: 'false',
+            AUTH_REGISTER_IP_LIMIT: '100',
+            AUTH_REGISTER_EMAIL_LIMIT: '100',
+        });
+        const created = await createApplication();
+        app = created.app;
+        database = app.get<Database>(DATABASE);
+        await app.listen(0, '127.0.0.1');
+        baseUrl = `http://127.0.0.1:${(app.getHttpServer().address() as AddressInfo).port}`;
+    });
+    afterAll(async () => {
+        await database.workspaceTransaction(workspaceId, async (tx) => {
+            await tx.query(
+                'DELETE FROM atlas.idempotency_requests WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.outbox_events WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.operations WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.collection_asset_transfers WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.product_variants WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.products WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.collection_items WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.extension_authorization_codes WHERE workspace_id = $1',
+                [workspaceId],
+            );
+            await tx.query(
+                'DELETE FROM atlas.extension_sessions WHERE workspace_id = $1',
+                [workspaceId],
+            );
+        });
+        await database.query(
+            'DELETE FROM atlas.workspace_memberships WHERE user_id = $1',
+            [userId],
+        );
+        await database.query('DELETE FROM atlas.workspaces WHERE id = $1', [
+            workspaceId,
+        ]);
+        await database.query('DELETE FROM atlas.users WHERE id = $1', [userId]);
+        await app.close();
+    });

    it('未登录令牌返回可操作的扩展会话错误', async () => {
        const session = await request('/api/extension/session');
        assert.equal(session.response.status, 401);
        assert.equal(session.body.error.code, 'SESSION_EXPIRED');
-        assert.equal(session.body.error.message, 'Atlas ERP 登录已过期，请重新登录');
+        assert.equal(
+            session.body.error.message,
+            'Atlas ERP 登录已过期，请重新登录',
+        );
    });

    it('通过 Atlas Web access token 写入来源商品，并以新快照覆盖重复采集', async () => {
-        const registration = await request('/api/auth/register', { method: 'POST', body: JSON.stringify({ email: `extension-${randomUUID()}@example.test`, password: 'correct-horse-battery-staple-123', password_confirmation: 'correct-horse-battery-staple-123' }) });
-        assert.equal(registration.response.status, 201); userId = registration.body.data.user.id; workspaceId = registration.body.data.workspace.id; const webAccessToken = registration.body.data.access_token as string;
-        const session = await request('/api/extension/session', { headers: { Authorization: `Bearer ${webAccessToken}` } }); assert.equal(session.response.status, 200); assert.equal(session.body.data.workspace.id, workspaceId);
-        const payload = { source: { platform: 'ozon', itemKey: '123456789', canonicalUrl: 'https://www.ozon.ru/product/demo-123456789/', collectedAt: new Date().toISOString(), parserVersion: 'test-1' }, common: { title: { value: '来源商品', confidence: 'verified' }, sourcePrice: { amount: '99.90', currencyCode: 'RUB' }, attributes: [], categories: [] }, variantAxes: [], variants: [{ sourceVariantKey: '123456789', sourceSku: '123456789', displayName: '默认 SKU', selections: [], confidence: 'verified', evidenceRefs: ['page'] }], assets: [{ url: 'https://images.example.test/product.jpg', role: 'source_main', relation: 'product_candidate', sortOrder: 0, evidenceRef: 'page' }], evidence: [{ ref: 'page', kind: 'page', path: 'test' }] };
-        const first = await request('/api/extension/materials', { method: 'POST', headers: { Authorization: `Bearer ${webAccessToken}`, 'Idempotency-Key': randomUUID() }, body: JSON.stringify(payload) }); assert.equal(first.response.status, 201); assert.equal(first.body.data.reused, false); assert.equal(first.body.data.materialization.queued, 1);
+        const registration = await request('/api/auth/register', {
+            method: 'POST',
+            body: JSON.stringify({
+                email: `extension-${randomUUID()}@example.test`,
+                password: 'correct-horse-battery-staple-123',
+                password_confirmation: 'correct-horse-battery-staple-123',
+            }),
+        });
+        assert.equal(registration.response.status, 201);
+        userId = registration.body.data.user.id;
+        workspaceId = registration.body.data.workspace.id;
+        const webAccessToken = registration.body.data.access_token as string;
+        const session = await request('/api/extension/session', {
+            headers: { Authorization: `Bearer ${webAccessToken}` },
+        });
+        assert.equal(session.response.status, 200);
+        assert.equal(session.body.data.workspace.id, workspaceId);
+        const payload = {
+            source: {
+                platform: 'ozon',
+                itemKey: '123456789',
+                canonicalUrl: 'https://www.ozon.ru/product/demo-123456789/',
+                collectedAt: new Date().toISOString(),
+                parserVersion: 'test-1',
+            },
+            common: {
+                title: { value: '来源商品', confidence: 'verified' },
+                sourcePrice: { amount: '99.90', currencyCode: 'RUB' },
+                attributes: [],
+                categories: [],
+            },
+            variantAxes: [],
+            variants: [
+                {
+                    sourceVariantKey: '123456789',
+                    sourceSku: '123456789',
+                    displayName: '默认 SKU',
+                    selections: [],
+                    confidence: 'verified',
+                    evidenceRefs: ['page'],
+                },
+            ],
+            assets: [
+                {
+                    url: 'https://images.example.test/product.jpg',
+                    role: 'source_main',
+                    relation: 'product_candidate',
+                    sortOrder: 0,
+                    evidenceRef: 'page',
+                },
+            ],
+            evidence: [{ ref: 'page', kind: 'page', path: 'test' }],
+        };
+        const first = await request('/api/extension/materials', {
+            method: 'POST',
+            headers: {
+                Authorization: `Bearer ${webAccessToken}`,
+                'Idempotency-Key': randomUUID(),
+            },
+            body: JSON.stringify(payload),
+        });
+        assert.equal(first.response.status, 201);
+        assert.equal(first.body.data.reused, false);
+        assert.equal(first.body.data.collection_item.status, 'ready_for_edit');
+        assert.equal(first.body.data.materialization.pending_selection, 1);
+        assert.equal(first.body.data.materialization.queued, 0);
        // A sibling Ozon URL has a different entered SKU. It must still reuse
        // the item because the submitted real-SKU sets overlap at 123456789.
-        const refreshedPayload = { ...payload, source: { ...payload.source, itemKey: '987654321', canonicalUrl: 'https://www.ozon.ru/product/demo-987654321/', collectedAt: new Date(Date.now() + 1_000).toISOString(), parserVersion: 'test-2' }, common: { ...payload.common, title: { value: '第二次采集的中文标题', confidence: 'verified' }, sourcePrice: { amount: '109.90', currencyCode: 'RUB' } }, variants: [{ sourceVariantKey: '123456789', sourceSku: '123456789', displayName: '第一 SKU', selections: [], confidence: 'verified', evidenceRefs: ['page'] }, { sourceVariantKey: '987654321', sourceSku: '987654321', displayName: '第二次采集 SKU', selections: [], confidence: 'verified', evidenceRefs: ['page'] }], assets: [{ url: 'https://images.example.test/refreshed-product.jpg', role: 'source_main', relation: 'product_candidate', sortOrder: 0, evidenceRef: 'page' }] };
-        const second = await request('/api/extension/materials', { method: 'POST', headers: { Authorization: `Bearer ${webAccessToken}`, 'Idempotency-Key': randomUUID() }, body: JSON.stringify(refreshedPayload) }); assert.equal(second.response.status, 200); assert.equal(second.body.data.reused, true); assert.equal(second.body.data.refreshed, true); assert.equal(second.body.data.collection_item.id, first.body.data.collection_item.id); assert.equal(second.body.data.materialization.queued, 1);
-        const inspection = await request(`/api/collection-items/${first.body.data.collection_item.id}`, { headers: { Authorization: `Bearer ${webAccessToken}` } }); assert.equal(inspection.response.status, 200); assert.equal(inspection.body.data.title, '第二次采集的中文标题'); assert.equal(inspection.body.data.normalized_payload.source.parserVersion, 'test-2'); assert.equal(inspection.body.data.variants.length, 2); assert.deepEqual(inspection.body.data.variants.map((variant: { source_variant_key: string }) => variant.source_variant_key), ['123456789', '987654321']); assert.equal(inspection.body.data.asset_transfers.length, 1); assert.equal(inspection.body.data.asset_transfers[0].source_url, 'https://images.example.test/refreshed-product.jpg');
+        const refreshedPayload = {
+            ...payload,
+            source: {
+                ...payload.source,
+                itemKey: '987654321',
+                canonicalUrl: 'https://www.ozon.ru/product/demo-987654321/',
+                collectedAt: new Date(Date.now() + 1_000).toISOString(),
+                parserVersion: 'test-2',
+            },
+            common: {
+                ...payload.common,
+                title: {
+                    value: '第二次采集的中文标题',
+                    confidence: 'verified',
+                },
+                sourcePrice: { amount: '109.90', currencyCode: 'RUB' },
+            },
+            variants: [
+                {
+                    sourceVariantKey: '123456789',
+                    sourceSku: '123456789',
+                    displayName: '第一 SKU',
+                    selections: [],
+                    confidence: 'verified',
+                    evidenceRefs: ['page'],
+                },
+                {
+                    sourceVariantKey: '987654321',
+                    sourceSku: '987654321',
+                    displayName: '第二次采集 SKU',
+                    selections: [],
+                    confidence: 'verified',
+                    evidenceRefs: ['page'],
+                },
+            ],
+            assets: [
+                {
+                    url: 'https://images.example.test/refreshed-product.jpg',
+                    role: 'source_main',
+                    relation: 'product_candidate',
+                    sortOrder: 0,
+                    evidenceRef: 'page',
+                },
+                {
+                    url: 'https://videos.example.test/refreshed-product.mp4',
+                    mediaType: 'video',
+                    posterUrl: 'https://images.example.test/refreshed-video-poster.jpg',
+                    role: 'source_main',
+                    relation: 'product_candidate',
+                    sortOrder: 1,
+                    evidenceRef: 'page',
+                },
+            ],
+        };
+        const second = await request('/api/extension/materials', {
+            method: 'POST',
+            headers: {
+                Authorization: `Bearer ${webAccessToken}`,
+                'Idempotency-Key': randomUUID(),
+            },
+            body: JSON.stringify(refreshedPayload),
+        });
+        assert.equal(second.response.status, 200);
+        assert.equal(second.body.data.reused, true);
+        assert.equal(second.body.data.refreshed, true);
+        assert.equal(
+            second.body.data.collection_item.id,
+            first.body.data.collection_item.id,
+        );
+        assert.equal(second.body.data.materialization.pending_selection, 2);
+        assert.equal(second.body.data.materialization.queued, 0);
+        const inspection = await request(
+            `/api/collection-items/${first.body.data.collection_item.id}`,
+            { headers: { Authorization: `Bearer ${webAccessToken}` } },
+        );
+        assert.equal(inspection.response.status, 200);
+        assert.equal(inspection.body.data.title, '第二次采集的中文标题');
+        assert.equal(
+            inspection.body.data.normalized_payload.source.parserVersion,
+            'test-2',
+        );
+        assert.equal(inspection.body.data.variants.length, 2);
+        assert.deepEqual(
+            inspection.body.data.variants.map(
+                (variant: { source_variant_key: string }) =>
+                    variant.source_variant_key,
+            ),
+            ['123456789', '987654321'],
+        );
+        assert.equal(inspection.body.data.asset_transfers.length, 2);
+        assert.equal(
+            inspection.body.data.asset_transfers[0].source_url,
+            'https://images.example.test/refreshed-product.jpg',
+        );
+        assert.equal(
+            inspection.body.data.asset_transfers[0].status,
+            'pending_selection',
+        );
+        const videoTransfer = inspection.body.data.asset_transfers.find(
+            (transfer: { media_type: string }) => transfer.media_type === 'video',
+        );
+        assert.equal(videoTransfer?.source_url, 'https://videos.example.test/refreshed-product.mp4');
+        assert.equal(videoTransfer?.poster_url, 'https://images.example.test/refreshed-video-poster.jpg');
+        const eventsBeforeSelection = await database.workspaceTransaction(
+            workspaceId,
+            async (tx) =>
+                tx.query<{ count: string }>(
+                    `SELECT count(*)::text AS count FROM atlas.outbox_events
+                     WHERE event_type = 'collection.assets.materialize.requested'
+                       AND event_payload->>'collection_item_id' = $1`,
+                    [first.body.data.collection_item.id],
+                ),
+        );
+        assert.equal(eventsBeforeSelection.rows[0]?.count, '0');
+        const materialization = await request(
+            `/api/collection-items/${first.body.data.collection_item.id}/asset-materializations`,
+            {
+                method: 'POST',
+                headers: {
+                    Authorization: `Bearer ${webAccessToken}`,
+                    'Idempotency-Key': randomUUID(),
+                },
+                body: JSON.stringify({
+                    transfer_ids: [inspection.body.data.asset_transfers[0].id],
+                }),
+            },
+        );
+        assert.equal(materialization.response.status, 202);
+        const afterSelection = await request(
+            `/api/collection-items/${first.body.data.collection_item.id}`,
+            { headers: { Authorization: `Bearer ${webAccessToken}` } },
+        );
+        assert.equal(afterSelection.response.status, 200);
+        assert.ok(
+            ['queued', 'running', 'failed', 'succeeded'].includes(
+                afterSelection.body.data.asset_transfers[0].status,
+            ),
+        );
+        const eventsAfterSelection = await database.workspaceTransaction(
+            workspaceId,
+            async (tx) =>
+                tx.query<{ count: string }>(
+                    `SELECT count(*)::text AS count FROM atlas.outbox_events
+                     WHERE event_type = 'collection.assets.materialize.requested'
+                       AND event_payload->>'collection_item_id' = $1`,
+                    [first.body.data.collection_item.id],
+                ),
+        );
+        assert.equal(eventsAfterSelection.rows[0]?.count, '1');
+        const transferWorkspace = await database.workspaceTransaction(
+            workspaceId,
+            async (tx) =>
+                tx.query<{ workspace_id: string }>(
+                    'SELECT workspace_id FROM atlas.collection_asset_transfers WHERE id = $1',
+                    [inspection.body.data.asset_transfers[0].id],
+                ),
+        );
+        assert.equal(transferWorkspace.rows[0]?.workspace_id, workspaceId);
    });
 });