feat(edit): 编辑页上报ozon 开发

apps/worker/src/collection-assets.ts

@@ -3,6 +3,7 @@ import { lookup } from 'node:dns/promises';
 import { isIP } from 'node:net';
 import { CreateBucketCommand, HeadBucketCommand, PutObjectCommand, type S3Client } from '@aws-sdk/client-s3';
 import type { Database } from '@atlas/database';
+import { workerConfig } from './config.js';
 import type { OutboxJob } from './outbox-dispatcher.js';
 
 const MAX_IMAGE_BYTES = 20 * 1024 * 1024;
@@ -45,7 +46,7 @@ async function materializeOne(database: Database, storage: S3Client, bucket: str
                 return;
             }
             const asset = await tx.query<{ id: string }>(`INSERT INTO atlas.media_assets (workspace_id, storage_provider, bucket, object_key, content_type, byte_size, checksum_sha256, width_px, height_px)
-                VALUES (current_setting('app.workspace_id')::uuid, 'local_minio', $1, $2, $3, $4, $5, $6, $7) RETURNING id`, [bucket, objectKey, downloaded.contentType, downloaded.bytes.byteLength, checksum, downloaded.width, downloaded.height]);
+                VALUES (current_setting('app.workspace_id')::uuid, $8, $1, $2, $3, $4, $5, $6, $7) RETURNING id`, [bucket, objectKey, downloaded.contentType, downloaded.bytes.byteLength, checksum, downloaded.width, downloaded.height, workerConfig.objectStorageProvider]);
             await tx.query(`UPDATE atlas.collection_asset_transfers SET status = 'succeeded', asset_id = $2, completed_at = now(), updated_at = now(), error_payload = NULL WHERE id = $1 AND status = 'running' AND source_snapshot_revision = $3`, [transfer.id, asset.rows[0].id, snapshotRevision]);
         });
     } catch (error) {

apps/worker/src/config.ts

@@ -31,6 +31,11 @@ export const workerConfig = {
     databaseUrl: required('DATABASE_URL'),
     redisUrl: required('REDIS_URL'),
     storeSecretEncryptionKey: required('STORE_SECRET_ENCRYPTION_KEY'),
+    objectStorageProvider: process.env.OBJECT_STORAGE_PROVIDER ?? 'local_minio',
+    ozonImageDownloadOrigin: process.env.OZON_IMAGE_DOWNLOAD_ORIGIN ?? '',
+    ozonImageUrlTtlSeconds: Number(process.env.OZON_IMAGE_URL_TTL_SECONDS ?? 604800),
+    qiniuAccessKey: process.env.QINIU_ACCESS_KEY ?? process.env.OBJECT_STORAGE_ACCESS_KEY ?? '',
+    qiniuSecretKey: process.env.QINIU_SECRET_KEY ?? process.env.OBJECT_STORAGE_SECRET_KEY ?? '',
     objectStorageEndpoint:
         process.env.OBJECT_STORAGE_ENDPOINT ?? 'http://localhost:9000',
     objectStorageBucket: process.env.OBJECT_STORAGE_BUCKET ?? 'atlas-assets',

apps/worker/src/main.ts

@@ -29,6 +29,7 @@ import {
     markPlatformCatalogUnavailable,
     syncPlatformCatalog,
 } from './platform-catalog-sync.js';
+import { runProductPublish } from './product-publish.js';
 import { runProductAiTask } from './product-ai.js';
 import { markAspectRatioDerivativeFailed, runAspectRatioDerivative } from './media-aspect-ratio.js';
 import { markWatermarkDerivativeFailed, runWatermarkDerivative } from './media-watermark.js';
@@ -85,6 +86,8 @@ const worker = new Worker<OutboxJob | PlatformCatalogJob>(
     async (job) => {
         if (job.name === 'platform_catalog_sync.requested') {
             await syncPlatformCatalog(database, job.data as PlatformCatalogJob);
+        } else if (job.name === 'product_publish.requested') {
+            await runProductPublish(database, objectStorage, job.data as OutboxJob);
         } else if (job.name === 'product_ai_task.requested') {
             await runProductAiTask(database, job.data as OutboxJob);
         } else if (job.name === 'media.aspect_ratio_derivative.requested') {

apps/worker/src/online-product-mapper.ts

@@ -50,7 +50,8 @@ export function parseOzonListPage(value: unknown): {
 }
 
 export function parseOzonInfoItems(value: unknown): OzonProductDetail[] {
-    const result = record(record(value)?.result);
+    const root = record(value);
+    const result = Array.isArray(root?.items) ? root : record(root?.result);
     if (!result || !Array.isArray(result.items))
         throw new Error('Ozon 商品详情响应格式异常');
     return result.items.filter((item): item is OzonProductDetail =>

apps/worker/src/ozon-client.ts

@@ -11,6 +11,10 @@ export class OzonRequestError extends Error {
     }
 }
 export class OzonRetryableError extends Error {}
+/** 429 明确未受理写请求，可按平台要求延迟重试。 */
+export class OzonRateLimitError extends OzonRetryableError {
+    constructor(public readonly retryAfterSeconds: number) { super('Ozon request rate limited'); }
+}
 
 export type OzonCategoryNode = {
     /** 类型叶子会省略这两个字段，继承最近的父类目。 */
@@ -193,7 +197,13 @@ export async function requestOzon(
     if (response.status === 401 || response.status === 403) {
         throw new OzonCredentialError(response.status);
     }
-    if (response.status === 429 || response.status >= 500) {
+    if (response.status === 429) {
+        const header = response.headers.get('Item-Retry-After') ?? response.headers.get('Retry-After');
+        const parsed = Number(header);
+        const date = header ? (Date.parse(header)-Date.now())/1000 : NaN;
+        throw new OzonRateLimitError(Number.isFinite(parsed) && parsed > 0 ? Math.min(86400,Math.ceil(parsed)) : Number.isFinite(date) && date > 0 ? Math.min(86400,Math.ceil(date)) : 60);
+    }
+    if (response.status >= 500) {
         throw new OzonRetryableError(`Ozon temporarily unavailable (${response.status})`);
     }
     throw new OzonRequestError(response.status);

apps/worker/src/ozon-product-import.ts

@@ -16,10 +16,8 @@ export type OzonProductImportItem = {
     name: string;
     description_category_id: number;
     type_id: number;
-    price: number;
+    price: string;
     currency_code?: string;
-    description?: string;
-    barcode?: string;
     images?: string[];
     attributes: OzonProductImportAttribute[];
     depth: number;
@@ -78,6 +76,7 @@ export function buildOzonProductImportPayload(
     );
     const typeId = positiveInteger(source.category.type_id, 'category.type_id');
     const name = requiredText(source.title, 'title');
+    if ([...name].length > 500 || source.variants.length > 50) throw new OzonProductImportPayloadError('profile', 'title', '标题最多 500 字符，每批最多 50 个 SKU');
 
     return {
         items: source.variants.map((variant) => {
@@ -86,8 +85,7 @@ export function buildOzonProductImportPayload(
                 variant.package_spec,
             );
             const currencyCode = optionalText(variant.currency_code);
-            const description = optionalText(source.description);
-            const barcode = optionalText(source.barcode);
+            if ([...(variant.offer_id ?? '')].length > 50 || variant.images.length > 30) throw new OzonProductImportPayloadError(variant.variant_id, 'offer_id', '货号最多 50 字符，每 SKU 最多 30 张图片');
             return {
                 offer_id: requiredText(
                     variant.offer_id,
@@ -97,14 +95,8 @@ export function buildOzonProductImportPayload(
                 name,
                 description_category_id: categoryId,
                 type_id: typeId,
-                price: positiveDecimal(
-                    variant.price,
-                    'price',
-                    variant.variant_id,
-                ),
+                price: priceString(variant.price, variant.variant_id),
                 ...(currencyCode ? { currency_code: currencyCode } : {}),
-                ...(description ? { description } : {}),
-                ...(barcode ? { barcode } : {}),
                 ...(variant.images.length ? { images: variant.images } : {}),
                 // 每个 Ozon items[] 同时携带商品级属性和本 SKU 的 aspect 属性。
                 attributes: [
@@ -195,3 +187,9 @@ function positiveDecimal(
         `${field} 必须大于 0`,
     );
 }
+
+/** 售价保留十进制字符串，不用浮点数重新格式化报价快照。 */
+function priceString(value: string | null, variantId: string) {
+    positiveDecimal(value, 'price', variantId);
+    return value!.trim();
+}

apps/worker/src/ozon-publish-media.ts

@@ -0,0 +1,132 @@
+import { createHash, createHmac } from 'node:crypto';
+import {
+    GetObjectCommand,
+    PutObjectCommand,
+    type S3Client,
+} from '@aws-sdk/client-s3';
+import sharp from 'sharp';
+import type { OzonPublishItemSnapshot } from '@atlas/contracts';
+
+export type PublishMediaConfig = {
+    provider: string;
+    bucket: string;
+    downloadOrigin: string;
+    accessKey: string;
+    secretKey: string;
+    ttlSeconds: number;
+};
+
+/** 七牛私有下载凭证签署包含过期时间的完整 URL，凭证只在提交期间保留。 */
+export function qiniuPrivateDownloadUrl(
+    origin: string,
+    key: string,
+    deadline: number,
+    accessKey: string,
+    secretKey: string,
+): string {
+    const base = new URL(origin);
+    if (
+        base.protocol !== 'https:' ||
+        base.username ||
+        base.password ||
+        base.search ||
+        base.hash ||
+        base.pathname !== '/' ||
+        !accessKey ||
+        !secretKey ||
+        !Number.isSafeInteger(deadline) ||
+        deadline <= 0
+    )
+        throw new Error('请配置七牛公网 HTTPS 下载域名及签名凭据');
+    const url = `${base.origin}/${key.split('/').map(encodeURIComponent).join('/')}?e=${deadline}`;
+    const signature = createHmac('sha1', secretKey)
+        .update(url)
+        .digest('base64')
+        .replace(/\+/g, '-')
+        .replace(/\//g, '_');
+    return `${url}&token=${accessKey}:${signature}`;
+}
+
+/** 为本次发布复制不可变图片，源 checksum 不符时禁止交付；WebP 转为 JPEG。 */
+export async function prepareOzonImages(
+    storage: S3Client,
+    snapshot: OzonPublishItemSnapshot,
+    workspaceId: string,
+    jobId: string,
+    config: PublishMediaConfig,
+): Promise<string[]> {
+    if (
+        config.provider !== 'qiniu_kodo' ||
+        !config.downloadOrigin ||
+        !Number.isSafeInteger(config.ttlSeconds) ||
+        config.ttlSeconds < 3600
+    )
+        throw new Error(
+            '上架图片交付尚未配置，请配置七牛 HTTPS 下载域名和有效期',
+        );
+    const urls: string[] = [];
+    for (const media of snapshot.media) {
+        if (
+            media.storage_provider !== config.provider ||
+            media.bucket !== config.bucket
+        )
+            throw new Error('图片存储环境与当前七牛配置不一致，请先迁移素材');
+        const object = await storage.send(
+            new GetObjectCommand({
+                Bucket: media.bucket,
+                Key: media.object_key,
+            }),
+            { abortSignal: AbortSignal.timeout(60_000) },
+        );
+        if (!object.Body) throw new Error('发布源图片不存在');
+        const chunks: Buffer[] = [];
+        let size = 0;
+        for await (const chunk of object.Body as AsyncIterable<Uint8Array>) {
+            size += chunk.byteLength;
+            if (size > 20 * 1024 * 1024) throw new Error('发布图片超过 20 MB');
+            chunks.push(Buffer.from(chunk));
+        }
+        const source = Buffer.concat(chunks);
+        if (
+            createHash('sha256').update(source).digest('hex') !==
+            media.checksum_sha256
+        )
+            throw new Error('图片内容已变更，请重新保存草稿后上架');
+        const image = sharp(source, {
+            limitInputPixels: 100_000_000,
+            failOn: 'error',
+        });
+        const meta = await image.metadata();
+        if (!['jpeg', 'png', 'webp'].includes(meta.format ?? ''))
+            throw new Error('发布图片实际格式不受支持');
+        const webp = meta.format === 'webp';
+        const bytes = webp
+            ? await image
+                  .rotate()
+                  .flatten({ background: '#ffffff' })
+                  .jpeg({ quality: 95 })
+                  .toBuffer()
+            : source;
+        const ext = meta.format === 'png' ? 'png' : 'jpg';
+        const key = `publishing/${workspaceId}/${jobId}/${media.asset_id}-${media.checksum_sha256}.${ext}`;
+        await storage.send(
+            new PutObjectCommand({
+                Bucket: media.bucket,
+                Key: key,
+                Body: bytes,
+                ContentType: ext === 'png' ? 'image/png' : 'image/jpeg',
+            }),
+            { abortSignal: AbortSignal.timeout(60_000) },
+        );
+        urls.push(
+            qiniuPrivateDownloadUrl(
+                config.downloadOrigin,
+                key,
+                Math.floor(Date.now() / 1000) + config.ttlSeconds,
+                config.accessKey,
+                config.secretKey,
+            ),
+        );
+    }
+    return urls;
+}

apps/worker/src/product-ai.ts

@@ -101,7 +101,7 @@ export async function runProductAiTask(database: Database, job: OutboxJob) {
                     taskId,
                     status,
                     JSON.stringify(generated.result),
-                    process.env.DEEPSEEK_MODEL_NAME || 'deepseek-chat',
+                    process.env.DEEPSEEK_MODEL_NAME || 'deepseek-flash',
                 ],
             ),
         );
@@ -605,6 +605,8 @@ function aiFailureMessage(code: string) {
             return 'DeepSeek 当前繁忙或限流，请稍后重试';
         case 'DEEPSEEK_EMPTY_RESULT':
             return 'DeepSeek 未返回生成内容，请稍后重试';
+        case 'DEEPSEEK_OUTPUT_TRUNCATED':
+            return 'DeepSeek 生成内容超过长度限制，请精简采集资料后重试';
         case 'DEEPSEEK_INVALID_JSON':
             return 'DeepSeek 返回格式异常，请稍后重试';
         case 'DEEPSEEK_UNAVAILABLE':
@@ -622,15 +624,16 @@ function aiFailureMessage(code: string) {
     }
 }
 
-/** 官方 JSON 模式可能返回空内容，因此解析与网络调用分别有限次重试。 */
-async function deepSeekJson(
+/** 结构化填报不需要思考输出；避免推理耗尽正文预算，并对空白和截断结果有限次重试。 */
+export async function deepSeekJson(
     system: string,
     user: string,
 ): Promise<ModelResponse> {
     const key = process.env.DEEPSEEK_API_KEY;
     if (!key) throw new AiProviderError('DEEPSEEK_NOT_CONFIGURED');
-    const model = process.env.DEEPSEEK_MODEL_NAME || 'deepseek-chat';
+    const model = process.env.DEEPSEEK_MODEL_NAME || 'deepseek-flash';
     let last: unknown;
+    let maxTokens = 8192;
     for (let attempt = 0; attempt < 3; attempt += 1) {
         try {
             const response = await fetch(
@@ -648,10 +651,17 @@ async function deepSeekJson(
                                 role: 'system',
                                 content: `${system}\n输出必须是 JSON。`,
                             },
-                            { role: 'user', content: user },
+                            {
+                                role: 'user',
+                                content:
+                                    attempt === 0
+                                        ? user
+                                        : `${user}\n请立即返回完整的 JSON 对象，不要只返回空白。`,
+                            },
                         ],
+                        thinking: { type: 'disabled' },
                         response_format: { type: 'json_object' },
-                        max_tokens: 3500,
+                        max_tokens: maxTokens,
                         temperature: 0.2,
                     }),
                     signal: AbortSignal.timeout(45_000),
@@ -663,10 +673,18 @@ async function deepSeekJson(
                 throw new AiProviderError('DEEPSEEK_REJECTED');
             }
             const body = (await response.json()) as {
-                choices?: Array<{ message?: { content?: string } }>;
+                choices?: Array<{
+                    finish_reason?: string;
+                    message?: { content?: string };
+                }>;
                 usage?: { total_tokens?: number };
             };
-            const content = body.choices?.[0]?.message?.content;
+            const choice = body.choices?.[0];
+            if (choice?.finish_reason === 'length') {
+                maxTokens = 16384;
+                throw new AiProviderError('DEEPSEEK_OUTPUT_TRUNCATED');
+            }
+            const content = choice?.message?.content?.trim();
             if (!content) throw new AiProviderError('DEEPSEEK_EMPTY_RESULT');
             const parsed = JSON.parse(content) as unknown;
             if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed))

apps/worker/src/product-publish.ts

@@ -0,0 +1,576 @@
+import { randomUUID } from 'node:crypto';
+import type {
+    OzonPublishItemSnapshot,
+    ProductPublishItemStatus,
+} from '@atlas/contracts';
+import type { Database, DatabaseTransaction } from '@atlas/database';
+import type { S3Client } from '@aws-sdk/client-s3';
+import type { OutboxJob } from './outbox-dispatcher.js';
+import { workerConfig } from './config.js';
+import { decryptStoreSecret } from './store-secret.crypto.js';
+import {
+    OzonCredentialError,
+    OzonRequestError,
+    OzonRateLimitError,
+    requestOzon,
+} from './ozon-client.js';
+import { prepareOzonImages } from './ozon-publish-media.js';
+
+type Item = {
+    id: string;
+    listing_id: string;
+    variant_id: string;
+    offer_id: string;
+    status: ProductPublishItemStatus;
+    snapshot: OzonPublishItemSnapshot;
+};
+type Job = {
+    id: string;
+    product_id: string;
+    profile_id: string;
+    store_id: string;
+    status: string;
+    ozon_task_id: string | null;
+    submit_started_at: string | null;
+    poll_count: number;
+};
+type ErrorInfo = { code: string; field?: string; message: string };
+export type PublishDependencies = {
+    request: typeof requestOzon;
+    images: (
+        snapshot: OzonPublishItemSnapshot,
+        workspaceId: string,
+        jobId: string,
+    ) => Promise<string[]>;
+    decrypt: (cipher: Buffer, nonce: Buffer) => string;
+};
+
+/** 发布输入按 SKU 冻结，VAT、条码、仓库、库存以及顶层描述均不进入首期请求。 */
+export function publishRequestItem(
+    snapshot: OzonPublishItemSnapshot,
+    images: string[],
+) {
+    const item = {
+        offer_id: snapshot.offer_id,
+        name: snapshot.name,
+        description_category_id: snapshot.description_category_id,
+        type_id: snapshot.type_id,
+        price: snapshot.price,
+        ...(snapshot.old_price ? { old_price: snapshot.old_price } : {}),
+        currency_code: snapshot.currency_code,
+        depth: snapshot.depth,
+        width: snapshot.width,
+        height: snapshot.height,
+        weight: snapshot.weight,
+        dimension_unit: snapshot.dimension_unit,
+        weight_unit: snapshot.weight_unit,
+        attributes: snapshot.attributes,
+    };
+    if (
+        !images.length ||
+        images.length > 30 ||
+        [...item.offer_id].length > 50 ||
+        [...item.name].length > 500
+    )
+        throw new Error('发布快照超出 Ozon 字段上限');
+    if (
+        ![item.depth, item.width, item.height, item.weight].every(
+            (n) => Number.isInteger(n) && n > 0,
+        )
+    )
+        throw new Error('包装规格必须为正整数');
+    return { ...item, images };
+}
+export function parseInfoProducts(
+    body: unknown,
+): Array<{ offer_id: string; id: string }> {
+    const data = body as { items?: unknown; result?: { items?: unknown } };
+    const items = data?.items ?? data?.result?.items;
+    if (!Array.isArray(items)) throw new Error('Ozon 商品详情响应格式异常');
+    return items.map((raw) => {
+        const item = raw as {
+            offer_id?: unknown;
+            id?: unknown;
+            product_id?: unknown;
+        };
+        const id = String(item.id ?? item.product_id ?? '');
+        if (
+            typeof item.offer_id !== 'string' ||
+            !/^\d+$/.test(id) ||
+            id === '0'
+        )
+            throw new Error('Ozon 商品详情缺少标识');
+        return { offer_id: item.offer_id, id };
+    });
+}
+export function safePublishErrors(input: unknown): ErrorInfo[] {
+    if (!Array.isArray(input)) return [];
+    return input.slice(0, 30).map((raw) => {
+        const item = raw as Record<string, unknown>;
+        return {
+            code: clean(item.code ?? 'OZON_IMPORT_FAILED', 100),
+            ...(typeof item.field === 'string'
+                ? { field: clean(item.field, 200) }
+                : {}),
+            message: clean(
+                item.message ??
+                    item.description ??
+                    'Ozon 拒绝该 SKU，请核对商品资料',
+                800,
+            ),
+        };
+    });
+}
+function clean(value: unknown, max: number) {
+    return String(value)
+        .replace(/https?:\/\/[^\s"<>]+/gi, '[链接已隐藏]')
+        .replace(
+            /(?:api[-_ ]?key|token|signature)\s*[:=]\s*[^\s,;]+/gi,
+            '[凭证已隐藏]',
+        )
+        .slice(0, max);
+}
+
+/** 一次事件执行一次提交或轮询；延迟由持久化 Outbox 控制，不长期占用 Worker。 */
+export async function runProductPublish(
+    database: Database,
+    storage: S3Client,
+    event: OutboxJob,
+    overrides?: Partial<PublishDependencies>,
+) {
+    const dependencies: PublishDependencies = {
+        request: requestOzon,
+        images: (snapshot, workspaceId, jobId) =>
+            prepareOzonImages(storage, snapshot, workspaceId, jobId, {
+                provider: workerConfig.objectStorageProvider,
+                bucket: workerConfig.objectStorageBucket,
+                downloadOrigin: workerConfig.ozonImageDownloadOrigin,
+                accessKey: workerConfig.qiniuAccessKey,
+                secretKey: workerConfig.qiniuSecretKey,
+                ttlSeconds: workerConfig.ozonImageUrlTtlSeconds,
+            }),
+        decrypt: (cipher, nonce) =>
+            decryptStoreSecret(
+                cipher,
+                nonce,
+                workerConfig.storeSecretEncryptionKey,
+            ),
+        ...overrides,
+    };
+    const id = String(event.payload.publish_job_id ?? '');
+    const token = randomUUID();
+    const claimed = await database.workspaceTransaction(
+        event.workspaceId,
+        async (tx) => {
+            const result = await tx.query<Job & { lease_until: string | null }>(
+                `SELECT * FROM atlas.product_publish_jobs WHERE id=$1 FOR UPDATE`,
+                [id],
+            );
+            const job = result.rows[0];
+            if (
+                !job ||
+                ['succeeded', 'partially_succeeded', 'failed'].includes(
+                    job.status,
+                )
+            )
+                return null;
+            if (
+                job.lease_until &&
+                new Date(job.lease_until).getTime() > Date.now()
+            ) {
+                await enqueue(
+                    tx,
+                    event.workspaceId,
+                    id,
+                    Math.ceil(
+                        (new Date(job.lease_until).getTime() - Date.now()) /
+                            1000,
+                    ) + 1,
+                );
+                return null;
+            }
+            await tx.query(
+                `UPDATE atlas.product_publish_jobs SET lease_token=$2,lease_until=now()+interval '30 minutes',status=CASE WHEN status='reconciling' THEN status ELSE 'running' END,updated_at=now() WHERE id=$1`,
+                [id, token],
+            );
+            const items = await tx.query<Item>(
+                `SELECT id,listing_id,variant_id,offer_id,status,snapshot FROM atlas.product_publish_job_items WHERE job_id=$1 ORDER BY created_at,id`,
+                [id],
+            );
+            return { job, items: items.rows };
+        },
+    );
+    if (!claimed) return;
+    const { job } = claimed;
+    let items = claimed.items.filter(
+        (i) => !['imported', 'failed', 'skipped'].includes(i.status),
+    );
+    const mutate = async <T>(work: (tx: DatabaseTransaction) => Promise<T>) =>
+        database.workspaceTransaction(event.workspaceId, async (tx) => {
+            // 与草稿保存采用相同锁顺序，成功写入与型号变更不能交错。
+            await tx.query(
+                'SELECT id FROM atlas.products WHERE id=$1 FOR UPDATE',
+                [job.product_id],
+            );
+            await tx.query(
+                'SELECT id FROM atlas.product_platform_profiles WHERE id=$1 FOR UPDATE',
+                [job.profile_id],
+            );
+            const current = await tx.query(
+                `SELECT id FROM atlas.product_publish_jobs WHERE id=$1 AND lease_token=$2 FOR UPDATE`,
+                [id, token],
+            );
+            if (!current.rows.length) throw new Error('发布任务处理权已转移');
+            return work(tx);
+        });
+    const release = async (delay: number, reconciling = false) =>
+        mutate(async (tx) => {
+            await tx.query(
+                `UPDATE atlas.product_publish_jobs SET status=$3,lease_until=NULL,lease_token=NULL,poll_count=poll_count+1,updated_at=now() WHERE id=$1 AND lease_token=$2`,
+                [id, token, reconciling ? 'reconciling' : 'running'],
+            );
+            if (reconciling)
+                await tx.query(
+                    `UPDATE atlas.product_publish_job_items SET status='reconciling',errors=$2::jsonb,updated_at=now() WHERE job_id=$1 AND status IN ('queued','submitting','polling','reconciling')`,
+                    [
+                        id,
+                        JSON.stringify([
+                            {
+                                code: 'OZON_SUBMISSION_UNCERTAIN',
+                                message: '平台提交结果待对账，请勿重复提交',
+                            },
+                        ]),
+                    ],
+                );
+            await enqueue(tx, event.workspaceId, id, delay);
+        });
+    const finish = () =>
+        mutate(async (tx) => {
+            const counts = await tx.query<{
+                total: string;
+                imported: string;
+                pending: string;
+            }>(
+                `SELECT count(*)::text AS total,count(*) FILTER(WHERE status='imported')::text AS imported,count(*) FILTER(WHERE status IN ('queued','submitting','polling','reconciling'))::text AS pending FROM atlas.product_publish_job_items WHERE job_id=$1`,
+                [id],
+            );
+            const c = counts.rows[0];
+            if (Number(c.pending)) return;
+            await tx.query(
+                `UPDATE atlas.product_publish_jobs SET status=$2,completed_at=now(),lease_until=NULL,lease_token=NULL,updated_at=now() WHERE id=$1`,
+                [
+                    id,
+                    Number(c.imported) === Number(c.total)
+                        ? 'succeeded'
+                        : Number(c.imported)
+                          ? 'partially_succeeded'
+                          : 'failed',
+                ],
+            );
+        });
+    const complete = async (
+        item: Item,
+        status: 'imported' | 'failed' | 'skipped',
+        productId: string | null,
+        errors: ErrorInfo[],
+    ) =>
+        mutate(async (tx) => {
+            await tx.query(
+                `UPDATE atlas.product_publish_job_items SET status=$2,platform_product_id=$3,errors=$4::jsonb,updated_at=now() WHERE id=$1 AND status NOT IN ('imported','failed','skipped')`,
+                [item.id, status, productId, JSON.stringify(errors)],
+            );
+            await tx.query(
+                `UPDATE atlas.product_variant_platform_listings SET status=$2,platform_product_id=CASE WHEN $2='imported' THEN $3 ELSE platform_product_id END,first_imported_at=CASE WHEN $2='imported' THEN coalesce(first_imported_at,now()) ELSE first_imported_at END,last_error=$4::jsonb,revision=revision+1,updated_at=now() WHERE id=$1 AND status='submitting'`,
+                [
+                    item.listing_id,
+                    status === 'imported' ? 'imported' : 'failed',
+                    productId,
+                    JSON.stringify(errors),
+                ],
+            );
+        });
+    try {
+        const credentials = await database.workspaceTransaction(
+            event.workspaceId,
+            async (tx) =>
+                (
+                    await tx.query<{
+                        platform_client_id: string;
+                        secret_ciphertext: Buffer;
+                        secret_nonce: Buffer;
+                    }>(
+                        `SELECT s.platform_client_id,secret.secret_ciphertext,secret.secret_nonce FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.store_id=s.id AND secret.kind='seller_api_key' AND secret.revoked_at IS NULL WHERE s.id=$1 AND s.deleted_at IS NULL AND s.status='active' AND s.credential_status='valid' ORDER BY secret.created_at DESC LIMIT 1`,
+                        [job.store_id],
+                    )
+                ).rows[0],
+        );
+        if (!credentials) throw new OzonCredentialError(401);
+        const apiKey = dependencies.decrypt(
+            credentials.secret_ciphertext,
+            credentials.secret_nonce,
+        );
+        const remote = async () =>
+            parseInfoProducts(
+                await dependencies.request(
+                    credentials.platform_client_id,
+                    apiKey,
+                    '/v3/product/info/list',
+                    { offer_id: items.map((i) => i.offer_id) },
+                ),
+            );
+        if (job.ozon_task_id) {
+            const body = (await dependencies.request(
+                credentials.platform_client_id,
+                apiKey,
+                '/v1/product/import/info',
+                { task_id: Number(job.ozon_task_id) },
+            )) as {
+                result?: {
+                    items?: Array<{
+                        offer_id: string;
+                        status: string;
+                        product_id?: number | string;
+                        errors?: unknown;
+                    }>;
+                };
+            };
+            if (!Array.isArray(body.result?.items))
+                throw new Error('Ozon 导入任务响应格式异常');
+            for (const item of items) {
+                const outcome = body.result.items.find(
+                    (r) => r.offer_id === item.offer_id,
+                );
+                if (!outcome) continue;
+                if (
+                    outcome.status === 'imported' &&
+                    outcome.product_id &&
+                    String(outcome.product_id) !== '0'
+                )
+                    await complete(
+                        item,
+                        'imported',
+                        String(outcome.product_id),
+                        [],
+                    );
+                else if (outcome.status === 'failed')
+                    await complete(
+                        item,
+                        'failed',
+                        null,
+                        safePublishErrors(outcome.errors).length
+                            ? safePublishErrors(outcome.errors)
+                            : [
+                                  {
+                                      code: 'OZON_IMPORT_FAILED',
+                                      message: 'Ozon 导入失败，请检查商品资料',
+                                  },
+                              ],
+                    );
+                else if (outcome.status === 'skipped') {
+                    const found = (await remote()).find(
+                        (r) => r.offer_id === item.offer_id,
+                    );
+                    if (found)
+                        await complete(item, 'imported', found.id, [
+                            {
+                                code: 'OZON_IMPORT_RECONCILED',
+                                message:
+                                    'Ozon 跳过该项，已通过商品信息对账确认货号存在',
+                            },
+                        ]);
+                    else
+                        await complete(item, 'skipped', null, [
+                            {
+                                code: 'OZON_IMPORT_SKIPPED',
+                                message:
+                                    'Ozon 跳过该 SKU，未确认导入成功，请检查平台记录',
+                            },
+                        ]);
+                }
+            }
+        } else if (job.submit_started_at) {
+            // 提交标记已写入但 task ID 未落库：只对账，绝不盲目重发导入。
+            const found = await remote();
+            for (const item of items) {
+                const match = found.find((r) => r.offer_id === item.offer_id);
+                if (match)
+                    await complete(item, 'imported', match.id, [
+                        {
+                            code: 'OZON_IMPORT_RECONCILED',
+                            message:
+                                '提交响应未保存，已通过货号对账确认商品存在',
+                        },
+                    ]);
+            }
+        } else {
+            const existing = await remote();
+            for (const item of items)
+                if (existing.some((r) => r.offer_id === item.offer_id))
+                    await complete(item, 'skipped', null, [
+                        {
+                            code: 'OZON_OFFER_ALREADY_EXISTS',
+                            message:
+                                '该货号已存在于当前店铺，本次未重复导入。请前往在线商品查看或在 Ozon 后台处理。',
+                        },
+                    ]);
+            items = items.filter(
+                (i) => !existing.some((r) => r.offer_id === i.offer_id),
+            );
+            if (!items.length) {
+                await finish();
+                return;
+            }
+            const payload = [];
+            for (const item of items) {
+                try {
+                    payload.push(
+                        publishRequestItem(
+                            item.snapshot,
+                            await dependencies.images(
+                                item.snapshot,
+                                event.workspaceId,
+                                id,
+                            ),
+                        ),
+                    );
+                } catch {
+                    // 本批素材全部准备成功才允许提交，存储原始错误不向用户暴露对象路径。
+                    for (const pending of items)
+                        await complete(pending, 'failed', null, [
+                            {
+                                code: 'PUBLISH_MEDIA_OR_SNAPSHOT_INVALID',
+                                message:
+                                    '发布图片或商品快照准备失败，请检查七牛交付配置、素材和包装规格后重试',
+                            },
+                        ]);
+                    await finish();
+                    return;
+                }
+                await mutate((tx) =>
+                    tx.query(
+                        `UPDATE atlas.product_publish_jobs SET lease_until=now()+interval '30 minutes' WHERE id=$1`,
+                        [id],
+                    ),
+                );
+            }
+            await mutate(async (tx) => {
+                await tx.query(
+                    `UPDATE atlas.product_publish_jobs SET submit_started_at=now(),updated_at=now() WHERE id=$1`,
+                    [id],
+                );
+                await tx.query(
+                    `UPDATE atlas.product_publish_job_items SET status='submitting',updated_at=now() WHERE job_id=$1 AND status='queued'`,
+                    [id],
+                );
+            });
+            let result: unknown;
+            try {
+                result = await dependencies.request(
+                    credentials.platform_client_id,
+                    apiKey,
+                    '/v3/product/import',
+                    { items: payload },
+                );
+            } catch (error) {
+                if (error instanceof OzonRateLimitError) {
+                    await mutate(async (tx) => {
+                        await tx.query(
+                            `UPDATE atlas.product_publish_jobs SET submit_started_at=NULL WHERE id=$1`,
+                            [id],
+                        );
+                        await tx.query(
+                            `UPDATE atlas.product_publish_job_items SET status='queued' WHERE job_id=$1 AND status='submitting'`,
+                            [id],
+                        );
+                    });
+                    await release(error.retryAfterSeconds);
+                    return;
+                }
+                if (
+                    error instanceof OzonRequestError ||
+                    error instanceof OzonCredentialError
+                ) {
+                    for (const item of items)
+                        await complete(item, 'failed', null, [
+                            {
+                                code: 'OZON_IMPORT_REJECTED',
+                                message: `Ozon 拒绝导入（HTTP ${error.status}），请检查店铺授权和商品资料`,
+                            },
+                        ]);
+                    await finish();
+                    return;
+                }
+                await release(30, true);
+                return;
+            }
+            const taskId = String(
+                (result as { result?: { task_id?: unknown } })?.result
+                    ?.task_id ?? '',
+            );
+            if (!/^\d+$/.test(taskId) || taskId === '0') {
+                await release(30, true);
+                return;
+            }
+            await mutate(async (tx) => {
+                await tx.query(
+                    `UPDATE atlas.product_publish_jobs SET ozon_task_id=$2 WHERE id=$1`,
+                    [id, taskId],
+                );
+                await tx.query(
+                    `UPDATE atlas.product_publish_job_items SET status='polling' WHERE job_id=$1 AND status='submitting'`,
+                    [id],
+                );
+            });
+        }
+        const pending = await mutate((tx) =>
+            tx.query(
+                `SELECT id FROM atlas.product_publish_job_items WHERE job_id=$1 AND status IN ('queued','submitting','polling','reconciling') LIMIT 1`,
+                [id],
+            ),
+        );
+        if (pending.rows.length)
+            await release(
+                job.poll_count > 120 ? 900 : 15,
+                Boolean(job.submit_started_at && !job.ozon_task_id) ||
+                    job.poll_count > 120,
+            );
+        else await finish();
+    } catch (error) {
+        // 读取或数据库失败仍保留平台状态，避免把未知结果当作可重发失败。
+        if (
+            !job.submit_started_at &&
+            !job.ozon_task_id &&
+            (error instanceof OzonCredentialError ||
+                error instanceof OzonRequestError)
+        ) {
+            for (const item of items)
+                await complete(item, 'failed', null, [
+                    {
+                        code: 'OZON_STORE_UNAVAILABLE',
+                        message: '店铺授权或读取权限不可用，请重新验证店铺',
+                    },
+                ]);
+            await finish();
+            return;
+        }
+        await release(
+            error instanceof OzonRateLimitError
+                ? error.retryAfterSeconds
+                : job.poll_count > 120
+                  ? 900
+                  : 60,
+            Boolean(job.submit_started_at && !job.ozon_task_id),
+        );
+    }
+}
+async function enqueue(
+    tx: DatabaseTransaction,
+    workspaceId: string,
+    id: string,
+    seconds: number,
+) {
+    await tx.query(
+        `INSERT INTO atlas.outbox_events(workspace_id,aggregate_type,aggregate_id,event_type,event_payload,next_attempt_at) VALUES($1,'product_publish_job',$2,'product_publish.requested',$3::jsonb,now()+make_interval(secs=>$4))`,
+        [workspaceId, id, JSON.stringify({ publish_job_id: id }), seconds],
+    );
+}