feat(category): 类目树、属性、属性字典维护，新增系统管理员管理类目数据

apps/worker/src/main.ts

@@ -19,17 +19,25 @@ import {
     type CatalogSyncScope,
 } from './ozon-catalog-sync.js';
 import { scheduleStoreCategoryTreeSync } from './store-category-tree-scheduler.js';
+import { PlatformCatalogDispatcher, type PlatformCatalogJob } from './platform-catalog-dispatcher.js';
+import { PlatformCatalogScheduler } from './platform-catalog-scheduler.js';
+import { markPlatformCatalogUnavailable, syncPlatformCatalog } from './platform-catalog-sync.js';
 
 const connection = new Redis(workerConfig.redisUrl, {
     maxRetriesPerRequest: null,
 });
 const database = createApplicationDatabase(workerConfig.databaseUrl);
 const queue = new Queue<OutboxJob>('atlas-dispatch', { connection });
+const platformQueue = new Queue<PlatformCatalogJob>('atlas-dispatch', { connection });
 const dispatcher = new OutboxDispatcher(
     database,
     queue,
     `worker_${randomUUID()}`,
 );
+const platformDispatcher = new PlatformCatalogDispatcher(
+    database, platformQueue, `platform_worker_${randomUUID()}`,
+);
+const platformScheduler = new PlatformCatalogScheduler(database);
 const objectStorage = new S3Client({
     endpoint: workerConfig.objectStorageEndpoint,
     region: 'us-east-1',
@@ -54,15 +62,17 @@ type CatalogSyncJob = OutboxJob & {
     };
 };
 
-const worker = new Worker<OutboxJob>(
+const worker = new Worker<OutboxJob | PlatformCatalogJob>(
     'atlas-dispatch',
     async (job) => {
-        if (job.name === 'store_credential_verification.requested') {
+        if (job.name === 'platform_catalog_sync.requested') {
+            await syncPlatformCatalog(database, job.data as PlatformCatalogJob);
+        } else if (job.name === 'store_credential_verification.requested') {
             await verifyStoreCredentials(database, job.data as OutboxJob & { payload: { store_id: string; operation_id: string; secret_id?: string } });
         } else if (job.name === 'collection.cleanup.requested') {
             await cleanupCollection(database, job.data as OutboxJob & { payload: { operation_id: string; collection_item_ids: string[] } });
         } else if (job.name === 'collection.assets.materialize.requested' || job.name === 'collection.assets.retry.requested') {
-            await materializeCollectionAssets(database, objectStorage, workerConfig.objectStorageBucket, job.data);
+            await materializeCollectionAssets(database, objectStorage, workerConfig.objectStorageBucket, job.data as OutboxJob);
         } else if (
             job.name === 'store_category_tree_sync.requested' ||
             job.name === 'store_category_attribute_sync.requested' ||
@@ -84,7 +94,9 @@ async function cleanupCollection(database: Database, job: OutboxJob & { payload:
 
 worker.on('failed', (job, error) => {
     if (!job || job.attemptsMade < (job.opts.attempts ?? 1)) return;
-    if (job.name === 'store_credential_verification.requested') {
+    if (job.name === 'platform_catalog_sync.requested') {
+        void markPlatformCatalogUnavailable(database, job.data as PlatformCatalogJob);
+    } else if (job.name === 'store_credential_verification.requested') {
         void markUnavailable(database, job.data as StoreVerificationJob, error);
     } else if (
         job.name === 'store_category_tree_sync.requested' ||
@@ -285,20 +297,36 @@ async function markUnavailable(
 
 let dispatcherTimer: ReturnType<typeof setInterval> | undefined;
 let mediaCleanupTimer: ReturnType<typeof setInterval> | undefined;
+let platformCatalogTimer: ReturnType<typeof setInterval> | undefined;
 async function bootstrap() {
     await queue.waitUntilReady();
     await worker.waitUntilReady();
     await dispatcher.dispatchBatch();
+    await platformDispatcher.dispatchBatch();
+    await platformScheduler.run();
     dispatcherTimer = setInterval(
-        () =>
+        () => {
             void dispatcher.dispatchBatch().catch((error: unknown) => {
                 console.error(
                     'outbox_dispatch_failed',
                     error instanceof Error ? error.message : error,
                 );
-            }),
+            });
+            void platformDispatcher.dispatchBatch().catch((error: unknown) => {
+                console.error(
+                    'platform_outbox_dispatch_failed',
+                    error instanceof Error ? error.message : error,
+                );
+            });
+        },
         2_000,
     );
+    platformCatalogTimer = setInterval(
+        () => void platformScheduler.run().catch((error: unknown) => {
+            console.error('platform_catalog_schedule_failed', error instanceof Error ? error.message : error);
+        }),
+        60_000,
+    );
     void mediaUploadCleanup.run().catch((error: unknown) => {
         console.error(
             'media_upload_cleanup_startup_failed',
@@ -329,8 +357,10 @@ void bootstrap().catch((error: unknown) => {
 async function shutdown() {
     if (dispatcherTimer) clearInterval(dispatcherTimer);
     if (mediaCleanupTimer) clearInterval(mediaCleanupTimer);
+    if (platformCatalogTimer) clearInterval(platformCatalogTimer);
     await worker.close();
     await queue.close();
+    await platformQueue.close();
     objectStorage.destroy();
     await database.close();
     await connection.quit();

apps/worker/src/ozon-client.ts

@@ -87,18 +87,85 @@ export async function fetchOzonDictionaryValues(
     },
 ) {
     const query = input.query?.trim() ?? '';
-    const path = query
-        ? '/v1/description-category/attribute/values/search'
-        : '/v1/description-category/attribute/values';
-    const body = await requestOzon(clientId, apiKey, path, {
+    const request = {
         attribute_id: numberOrText(input.attributeId),
         description_category_id: numberOrText(input.categoryId),
         type_id: numberOrText(input.typeId),
         language: 'ZH_HANS',
-        limit: query ? 100 : 1_000,
-        ...(query ? { value: query } : { last_value_id: 0 }),
-    });
-    return readResultArray<OzonDictionaryValue>(body, '属性字典');
+    };
+    if (query) {
+        const body = await requestOzon(
+            clientId, apiKey, '/v1/description-category/attribute/values/search',
+            { ...request, limit: 100, value: query },
+        );
+        return readResultArray<OzonDictionaryValue>(body, '属性字典');
+    }
+    const values: OzonDictionaryValue[] = [];
+    const seen = new Set<string>();
+    let lastValueId: number | string = 0;
+    // 大字典仅预取首批 1000 项；后续值由关键词搜索补充，重复游标不得造成无限循环。
+    for (let pageIndex = 0; pageIndex < 20 && values.length < 1_000; pageIndex += 1) {
+        const body = await requestOzon(
+            clientId, apiKey, '/v1/description-category/attribute/values',
+            { ...request, limit: 1_000 - values.length, last_value_id: lastValueId },
+        );
+        const page = readResultArray<OzonDictionaryValue>(body, '属性字典');
+        let added = 0;
+        for (const value of page) {
+            const id = String(value.id ?? '');
+            if (!id || seen.has(id)) continue;
+            seen.add(id);
+            values.push(value);
+            added += 1;
+        }
+        const nextId = page.at(-1)?.id;
+        if (
+            (body as { has_next?: unknown }).has_next !== true || !page.length ||
+            !added || nextId == null || String(nextId) === String(lastValueId)
+        ) break;
+        lastValueId = typeof nextId === 'string' ? numberOrText(nextId) : nextId;
+    }
+    return values;
+}
+
+/** 平台级字典按页获取，明确返回完整性；大字典不会伪装成已全量同步。 */
+export async function fetchOzonDictionaryBatch(
+    clientId: string,
+    apiKey: string,
+    input: { attributeId: string; categoryId: string; typeId: string },
+): Promise<{ values: OzonDictionaryValue[]; isComplete: boolean }> {
+    const values: OzonDictionaryValue[] = [];
+    const seen = new Set<string>();
+    let lastValueId: number | string = 0;
+    for (let pageIndex = 0; pageIndex < 5; pageIndex += 1) {
+        const body = await requestOzon(
+            clientId, apiKey, '/v1/description-category/attribute/values',
+            {
+                attribute_id: numberOrText(input.attributeId),
+                description_category_id: numberOrText(input.categoryId),
+                type_id: numberOrText(input.typeId),
+                language: 'ZH_HANS',
+                limit: 2000,
+                last_value_id: lastValueId,
+            },
+        );
+        const page = readResultArray<OzonDictionaryValue>(body, '属性字典');
+        for (const value of page) {
+            const id = String(value.id ?? '').trim();
+            if (id && !seen.has(id)) {
+                seen.add(id);
+                values.push(value);
+            }
+        }
+        const hasNext = (body as { has_next?: unknown }).has_next === true;
+        if (!hasNext) return { values, isComplete: true };
+        const nextId = page.at(-1)?.id;
+        if (!page.length || nextId == null || String(nextId) === String(lastValueId)) {
+            throw new OzonRetryableError('Ozon 字典分页游标没有前进');
+        }
+        lastValueId = nextId;
+    }
+    return { values, isComplete: false };
 }
 
 async function requestOzon(

apps/worker/src/platform-catalog-dispatcher.ts

@@ -0,0 +1,68 @@
+import type { Queue } from 'bullmq';
+import type { Database } from '@atlas/database';
+
+export type PlatformCatalogJob = {
+    eventId: string;
+    runId: string;
+    platformId: string;
+};
+
+export class PlatformCatalogDispatcher {
+    public constructor(
+        private readonly database: Database,
+        private readonly queue: Queue<PlatformCatalogJob>,
+        private readonly workerId: string,
+    ) {}
+
+    public async dispatchBatch() {
+        const events = await this.database.transaction((tx) => tx.query<{
+            id: string; run_id: string; platform_id: string;
+        }>(
+            `WITH candidates AS (
+                SELECT id FROM atlas.platform_catalog_outbox_events
+                WHERE (status = 'pending' AND next_attempt_at <= now())
+                   OR (status = 'dispatching' AND locked_until < now())
+                ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 20
+             )
+             UPDATE atlas.platform_catalog_outbox_events event
+             SET status = 'dispatching', locked_by = $1,
+                 locked_until = now() + interval '2 minutes', updated_at = now()
+             FROM candidates WHERE event.id = candidates.id
+             RETURNING event.id, event.run_id, event.platform_id`,
+            [this.workerId],
+        ));
+        for (const event of events.rows) {
+            try {
+                await this.queue.add('platform_catalog_sync.requested', {
+                    eventId: event.id,
+                    runId: event.run_id,
+                    platformId: event.platform_id,
+                }, {
+                    jobId: event.id,
+                    attempts: 5,
+                    backoff: { type: 'exponential', delay: 5_000 },
+                    removeOnComplete: 1000,
+                    removeOnFail: 1000,
+                });
+                await this.database.query(
+                    `UPDATE atlas.platform_catalog_outbox_events
+                     SET status = 'dispatched', dispatched_at = now(),
+                         locked_by = NULL, locked_until = NULL, updated_at = now()
+                     WHERE id = $1 AND status = 'dispatching' AND locked_by = $2`,
+                    [event.id, this.workerId],
+                );
+            } catch (error) {
+                await this.database.query(
+                    `UPDATE atlas.platform_catalog_outbox_events
+                     SET status = 'pending', attempt_count = attempt_count + 1,
+                         next_attempt_at = now() + interval '30 seconds',
+                         locked_by = NULL, locked_until = NULL,
+                         last_error_code = 'QUEUE_UNAVAILABLE', updated_at = now()
+                     WHERE id = $1 AND status = 'dispatching' AND locked_by = $2`,
+                    [event.id, this.workerId],
+                );
+                throw error;
+            }
+        }
+    }
+}

apps/worker/src/platform-catalog-mappers.ts

@@ -0,0 +1,107 @@
+import type { OzonCategoryAttribute, OzonCategoryNode } from './ozon-client.js';
+
+export type PlatformCategoryRecord = {
+    external_category_id: string;
+    external_type_id: string;
+    parent_key: string | null;
+    name: string;
+    path: string;
+    is_active: boolean;
+    source_payload: OzonCategoryNode;
+};
+
+/** Ozon 类型叶子可能省略类目 ID，需要继承最近的父类目。 */
+export function flattenPlatformCategories(
+    nodes: OzonCategoryNode[],
+    path: string[] = [],
+    parentKey: string | null = null,
+    inheritedCategory?: { id: string; name: string },
+): PlatformCategoryRecord[] {
+    const records: PlatformCategoryRecord[] = [];
+    for (const node of nodes) {
+        const explicitId = String(node.description_category_id ?? '').trim();
+        const explicitName = node.category_name?.trim() ?? '';
+        const categoryId = explicitId || inheritedCategory?.id || '';
+        const categoryName = explicitName || inheritedCategory?.name || '';
+        const typeId = String(node.type_id ?? 0);
+        const name = explicitName || node.type_name?.trim() || categoryName;
+        if (!categoryId || !name) continue;
+        const nextPath = [...path, name];
+        const key = `${categoryId}:${typeId}`;
+        records.push({
+            external_category_id: categoryId,
+            external_type_id: typeId,
+            parent_key: parentKey,
+            name,
+            path: nextPath.join(' › '),
+            is_active: node.disabled !== true && typeId !== '0',
+            source_payload: node,
+        });
+        records.push(...flattenPlatformCategories(
+            node.children ?? [],
+            nextPath,
+            key,
+            explicitId && explicitName ? { id: explicitId, name: explicitName } : inheritedCategory,
+        ));
+    }
+    return records;
+}
+
+export type PlatformAttributeRecord = {
+    platform_attribute_id: string;
+    complex_id: string;
+    name: string;
+    description: string | null;
+    group_id: string | null;
+    group_name: string | null;
+    value_type: 'text' | 'integer' | 'decimal' | 'boolean' | 'dictionary';
+    is_required: boolean;
+    is_collection: boolean;
+    max_value_count: number;
+    dictionary_id: string | null;
+    is_aspect: boolean;
+    category_dependent: boolean;
+    source_payload: OzonCategoryAttribute;
+};
+
+export function mapPlatformAttributes(values: OzonCategoryAttribute[]): PlatformAttributeRecord[] {
+    const byKey = new Map<string, PlatformAttributeRecord>();
+    for (const value of values) {
+        const attributeId = String(value.id ?? '').trim();
+        const name = value.name?.trim();
+        if (!attributeId || !name) continue;
+        const complexId = String(value.attribute_complex_id ?? value.complex_id ?? 0);
+        const dictionaryId = String(value.dictionary_id ?? '').trim();
+        const hasDictionary = !!dictionaryId && dictionaryId !== '0';
+        const isCollection = value.is_collection === true;
+        const requestedMax = Number(value.max_value_count ?? (isCollection ? 100 : 1));
+        const type = String(value.type ?? '').toLowerCase();
+        const record: PlatformAttributeRecord = {
+            platform_attribute_id: attributeId,
+            complex_id: complexId,
+            name,
+            description: stringOrNull(value.description),
+            group_id: value.group_id == null ? null : String(value.group_id),
+            group_name: stringOrNull(value.group_name),
+            value_type: hasDictionary ? 'dictionary'
+                : type === 'integer' ? 'integer'
+                    : type === 'decimal' || type === 'number' ? 'decimal'
+                        : type === 'boolean' ? 'boolean' : 'text',
+            is_required: value.is_required === true,
+            is_collection: isCollection,
+            max_value_count: isCollection
+                ? Math.max(1, Math.min(100, Number.isFinite(requestedMax) ? requestedMax : 100))
+                : 1,
+            dictionary_id: hasDictionary ? dictionaryId : null,
+            is_aspect: value.is_aspect === true,
+            category_dependent: value.category_dependent === true,
+            source_payload: value,
+        };
+        byKey.set(`${attributeId}:${complexId}`, record);
+    }
+    return [...byKey.values()];
+}
+
+function stringOrNull(value: unknown): string | null {
+    return typeof value === 'string' && value.trim() ? value.trim() : null;
+}

apps/worker/src/platform-catalog-scheduler.ts

@@ -0,0 +1,103 @@
+import type { Database, DatabaseTransaction } from '@atlas/database';
+
+type Candidate = { scope: 'tree' | 'attributes' | 'dictionary'; scope_key: string;
+    category_id: string | null; attribute_id: string | null };
+
+/** 定时任务逐平台限量排队，避免启动时一次扫完全部类目和大字典。 */
+export class PlatformCatalogScheduler {
+    public constructor(private readonly database: Database) {}
+
+    public async run() {
+        const sources = await this.database.query<{ platform_id: string }>(
+            `SELECT source.platform_id FROM atlas.platform_catalog_sources source
+             JOIN atlas.platforms platform ON platform.id = source.platform_id
+             WHERE platform.status = 'active' AND upper(platform.code) = 'OZON'`,
+        );
+        for (const source of sources.rows) {
+            await this.schedulePlatform(source.platform_id);
+        }
+    }
+
+    private async schedulePlatform(platformId: string) {
+        await this.database.transaction(async (tx) => {
+            const source = await tx.query<{ revision: number }>(
+                `SELECT revision FROM atlas.platform_catalog_sources
+                 WHERE platform_id = $1 FOR UPDATE`,
+                [platformId],
+            );
+            if (!source.rows[0]) return;
+            const pending = await tx.query<{ count: string }>(
+                `SELECT count(*)::text AS count FROM atlas.platform_catalog_sync_runs
+                 WHERE platform_id = $1 AND status IN ('queued', 'running')`,
+                [platformId],
+            );
+            if (Number(pending.rows[0]?.count ?? 0) >= 6) return;
+            const candidates = await tx.query<Candidate>(
+                `WITH possible AS (
+                    SELECT 'tree'::text AS scope, 'tree'::text AS scope_key,
+                           NULL::uuid AS category_id, NULL::uuid AS attribute_id,
+                           interval '24 hours' AS refresh_interval, 0 AS priority
+                    UNION ALL
+                    SELECT 'attributes', 'attributes:' || category.id, category.id, NULL::uuid,
+                           interval '24 hours', 1
+                    FROM atlas.platform_product_categories category
+                    WHERE category.platform_id = $1 AND category.is_active
+                      AND category.external_type_id IS NOT NULL AND category.external_type_id <> '0'
+                    UNION ALL
+                    SELECT 'dictionary', 'dictionary:' || attribute.id,
+                           attribute.platform_product_category_id, attribute.id,
+                           interval '7 days', 2
+                    FROM atlas.platform_product_category_attributes attribute
+                    JOIN atlas.platform_product_categories category
+                      ON category.id = attribute.platform_product_category_id
+                    WHERE attribute.platform_id = $1 AND attribute.is_active
+                      AND attribute.dictionary_id IS NOT NULL AND category.is_active
+                )
+                SELECT possible.scope, possible.scope_key, possible.category_id, possible.attribute_id
+                FROM possible
+                WHERE NOT EXISTS (
+                    SELECT 1 FROM atlas.platform_catalog_sync_runs run
+                    WHERE run.platform_id = $1 AND run.scope_key = possible.scope_key
+                      AND run.status = 'succeeded'
+                      AND run.completed_at > now() - possible.refresh_interval
+                ) AND NOT EXISTS (
+                    SELECT 1 FROM atlas.platform_catalog_sync_runs run
+                    WHERE run.platform_id = $1 AND run.scope_key = possible.scope_key
+                      AND run.created_at > now() - interval '1 hour'
+                ) AND NOT EXISTS (
+                    SELECT 1 FROM atlas.platform_catalog_sync_runs run
+                    WHERE run.platform_id = $1 AND run.scope_key = possible.scope_key
+                      AND run.status IN ('queued', 'running')
+                )
+                ORDER BY possible.priority, possible.scope_key LIMIT 6`,
+                [platformId],
+            );
+            const capacity = 6 - Number(pending.rows[0]?.count ?? 0);
+            for (const candidate of candidates.rows.slice(0, capacity)) {
+                await enqueueScheduled(tx, platformId, source.rows[0].revision, candidate);
+            }
+        });
+    }
+}
+
+async function enqueueScheduled(
+    tx: DatabaseTransaction,
+    platformId: string,
+    sourceRevision: number,
+    candidate: Candidate,
+) {
+    const run = await tx.query<{ id: string }>(
+        `INSERT INTO atlas.platform_catalog_sync_runs
+            (platform_id, scope, scope_key, category_id, attribute_id,
+             trigger_kind, source_revision)
+         VALUES ($1, $2, $3, $4, $5, 'scheduled', $6)
+         RETURNING id`,
+        [platformId, candidate.scope, candidate.scope_key,
+            candidate.category_id, candidate.attribute_id, sourceRevision],
+    );
+    await tx.query(
+        `INSERT INTO atlas.platform_catalog_outbox_events (run_id, platform_id)
+         VALUES ($1, $2)`,
+        [run.rows[0].id, platformId],
+    );
+}

apps/worker/src/platform-catalog-sync.ts

@@ -0,0 +1,337 @@
+import type { Database, DatabaseTransaction } from '@atlas/database';
+import {
+    fetchOzonCategoryAttributes, fetchOzonCategoryTree, fetchOzonDictionaryBatch,
+    OzonCredentialError, OzonRequestError, type OzonCategoryNode,
+    type OzonDictionaryValue,
+} from './ozon-client.js';
+import { flattenPlatformCategories, mapPlatformAttributes } from './platform-catalog-mappers.js';
+import type { PlatformCatalogJob } from './platform-catalog-dispatcher.js';
+import { decryptStoreSecret } from './store-secret.crypto.js';
+import { workerConfig } from './config.js';
+
+type RunSource = {
+    id: string; platform_id: string; scope: 'tree' | 'attributes' | 'dictionary';
+    category_id: string | null; attribute_id: string | null;
+    source_revision: number; platform_code: string;
+    client_id: string; secret_ciphertext: Buffer; secret_nonce: Buffer;
+    current_revision: number;
+    external_category_id: string | null; external_type_id: string | null;
+    platform_attribute_id: string | null;
+};
+
+type SyncPayload =
+    | { scope: 'tree'; nodes: OzonCategoryNode[] }
+    | { scope: 'attributes'; attributes: Awaited<ReturnType<typeof fetchOzonCategoryAttributes>> }
+    | { scope: 'dictionary'; batch: Awaited<ReturnType<typeof fetchOzonDictionaryBatch>> };
+
+export async function syncPlatformCatalog(database: Database, job: PlatformCatalogJob) {
+    const source = await claimRun(database, job);
+    if (!source) return;
+    if (source.source_revision !== source.current_revision) {
+        await finishRun(database, job.runId, 'cancelled', 'CATALOG_SOURCE_REPLACED', '同步来源已更新');
+        return;
+    }
+    if (source.platform_code.toUpperCase() !== 'OZON') {
+        await finishRun(database, job.runId, 'failed', 'PLATFORM_CATALOG_UNSUPPORTED', '该平台尚未接入目录同步');
+        return;
+    }
+    const apiKey = decryptStoreSecret(
+        source.secret_ciphertext, source.secret_nonce, workerConfig.storeSecretEncryptionKey,
+    );
+    try {
+        let payload: SyncPayload;
+        if (source.scope === 'tree') {
+            payload = { scope: 'tree', nodes: await fetchOzonCategoryTree(source.client_id, apiKey) };
+        } else {
+            if (!source.external_category_id || !source.external_type_id || source.external_type_id === '0') {
+                await finishRun(database, job.runId, 'failed', 'CATEGORY_TYPE_UNAVAILABLE', '商品类型不存在或已停用');
+                return;
+            }
+            if (source.scope === 'attributes') {
+                payload = {
+                    scope: 'attributes',
+                    attributes: await fetchOzonCategoryAttributes(
+                        source.client_id, apiKey,
+                        source.external_category_id, source.external_type_id,
+                    ),
+                };
+            } else {
+                if (!source.platform_attribute_id) {
+                    await finishRun(database, job.runId, 'failed', 'ATTRIBUTE_NOT_FOUND', '字典属性已失效');
+                    return;
+                }
+                payload = {
+                    scope: 'dictionary',
+                    batch: await fetchOzonDictionaryBatch(source.client_id, apiKey, {
+                        attributeId: source.platform_attribute_id,
+                        categoryId: source.external_category_id,
+                        typeId: source.external_type_id,
+                    }),
+                };
+            }
+        }
+        await persistAndComplete(database, source, payload);
+    } catch (error) {
+        if (error instanceof OzonCredentialError) {
+            await finishRun(
+                database, job.runId, 'failed', 'OZON_CATALOG_CREDENTIAL_INVALID',
+                error.status === 403 ? 'Ozon 拒绝读取目录，请检查密钥权限' : 'Ozon 未接受同步来源凭据',
+            );
+            return;
+        }
+        if (error instanceof OzonRequestError) {
+            await finishRun(
+                database, job.runId, 'failed', 'OZON_CATALOG_REQUEST_REJECTED',
+                `Ozon 拒绝目录同步请求（HTTP ${error.status}）`,
+            );
+            return;
+        }
+        throw error;
+    }
+}
+
+export async function markPlatformCatalogUnavailable(database: Database, job: PlatformCatalogJob) {
+    await finishRun(
+        database, job.runId, 'failed', 'PLATFORM_CATALOG_SYNC_UNAVAILABLE',
+        '暂时无法同步平台目录，请稍后重试',
+    );
+}
+
+async function claimRun(database: Database, job: PlatformCatalogJob): Promise<RunSource | null> {
+    return database.transaction(async (tx) => {
+        const result = await tx.query<RunSource>(
+            `SELECT run.id, run.platform_id, run.scope, run.category_id, run.attribute_id,
+                    run.source_revision, platform.code AS platform_code,
+                    source.client_id, source.secret_ciphertext, source.secret_nonce,
+                    source.revision AS current_revision,
+                    category.external_category_id, category.external_type_id,
+                    attribute.platform_attribute_id
+             FROM atlas.platform_catalog_sync_runs run
+             JOIN atlas.platforms platform ON platform.id = run.platform_id
+             JOIN atlas.platform_catalog_sources source ON source.platform_id = run.platform_id
+             LEFT JOIN atlas.platform_product_categories category
+               ON category.id = run.category_id AND category.platform_id = run.platform_id
+             LEFT JOIN atlas.platform_product_category_attributes attribute
+               ON attribute.id = run.attribute_id AND attribute.platform_id = run.platform_id
+             WHERE run.id = $1 AND run.platform_id = $2
+               AND run.status IN ('queued', 'running')
+             FOR UPDATE OF run`,
+            [job.runId, job.platformId],
+        );
+        const source = result.rows[0];
+        if (!source) return null;
+        await tx.query(
+            `UPDATE atlas.platform_catalog_sync_runs
+             SET status = 'running', started_at = coalesce(started_at, now()),
+                 attempt_count = attempt_count + 1
+             WHERE id = $1`,
+            [job.runId],
+        );
+        return source;
+    });
+}
+
+async function persistAndComplete(database: Database, source: RunSource, payload: SyncPayload) {
+    await database.transaction(async (tx) => {
+        // 凭据更新与结果发布共用来源行锁，旧任务不能覆盖新来源的目录。
+        const revision = await tx.query<{ revision: number }>(
+            `SELECT revision FROM atlas.platform_catalog_sources
+             WHERE platform_id = $1 FOR UPDATE`,
+            [source.platform_id],
+        );
+        const run = await tx.query<{ status: string }>(
+            `SELECT status FROM atlas.platform_catalog_sync_runs WHERE id = $1 FOR UPDATE`,
+            [source.id],
+        );
+        if (run.rows[0]?.status !== 'running') return;
+        if (revision.rows[0]?.revision !== source.source_revision) {
+            await updateRun(tx, source.id, 'cancelled', null, 'CATALOG_SOURCE_REPLACED', '同步来源已更新');
+            return;
+        }
+        let result: Record<string, unknown>;
+        if (payload.scope === 'tree') {
+            result = await persistTree(tx, source.platform_id, payload.nodes);
+        } else if (payload.scope === 'attributes') {
+            result = await persistAttributes(tx, source.platform_id, source.category_id ?? '', payload.attributes);
+        } else {
+            result = await persistDictionary(
+                tx, source.platform_id, source.category_id ?? '',
+                source.attribute_id ?? '', payload.batch.values, payload.batch.isComplete,
+            );
+        }
+        await updateRun(tx, source.id, 'succeeded', result);
+    });
+}
+
+async function persistTree(tx: DatabaseTransaction, platformId: string, nodes: OzonCategoryNode[]) {
+    const records = [...new Map(flattenPlatformCategories(nodes).map((node) => [
+        `${node.external_category_id}:${node.external_type_id}`, node,
+    ])).values()];
+    if (!records.length) throw new Error('Ozon 返回了空类目树');
+    const syncTime = new Date();
+    await tx.query(
+        `INSERT INTO atlas.platform_product_categories
+            (platform_id, external_category_id, external_type_id, name, path,
+             is_active, source_payload, synced_at)
+         SELECT $1, external_category_id, external_type_id, name, path,
+                is_active, source_payload, $3::timestamptz
+         FROM jsonb_to_recordset($2::jsonb) AS row(
+             external_category_id text, external_type_id text, parent_key text,
+             name text, path text, is_active boolean, source_payload jsonb)
+         ON CONFLICT (platform_id, external_category_id, external_type_id) DO UPDATE
+         SET name = EXCLUDED.name, path = EXCLUDED.path, is_active = EXCLUDED.is_active,
+             source_payload = EXCLUDED.source_payload, synced_at = EXCLUDED.synced_at,
+             updated_at = now()`,
+        [platformId, JSON.stringify(records), syncTime],
+    );
+    await tx.query(
+        `WITH incoming AS (
+            SELECT * FROM jsonb_to_recordset($2::jsonb) AS row(
+                external_category_id text, external_type_id text, parent_key text,
+                name text, path text, is_active boolean, source_payload jsonb)
+         ), resolved AS (
+            SELECT child.id AS child_id, parent.id AS parent_id
+            FROM incoming
+            JOIN atlas.platform_product_categories child
+              ON child.platform_id = $1
+             AND child.external_category_id = incoming.external_category_id
+             AND child.external_type_id = incoming.external_type_id
+            LEFT JOIN atlas.platform_product_categories parent
+              ON parent.platform_id = $1
+             AND parent.external_category_id || ':' || parent.external_type_id = incoming.parent_key
+         )
+         UPDATE atlas.platform_product_categories category
+         SET parent_id = resolved.parent_id, updated_at = now()
+         FROM resolved WHERE category.id = resolved.child_id`,
+        [platformId, JSON.stringify(records)],
+    );
+    await tx.query(
+        `UPDATE atlas.platform_product_categories
+         SET is_active = false, updated_at = now()
+         WHERE platform_id = $1 AND synced_at IS DISTINCT FROM $2::timestamptz`,
+        [platformId, syncTime],
+    );
+    return { category_count: records.length };
+}
+
+async function persistAttributes(
+    tx: DatabaseTransaction, platformId: string, categoryId: string,
+    attributes: Awaited<ReturnType<typeof fetchOzonCategoryAttributes>>,
+) {
+    const records = mapPlatformAttributes(attributes);
+    if (!records.length) throw new Error('Ozon 返回了空属性定义');
+    const syncTime = new Date();
+    await tx.query(
+        `INSERT INTO atlas.platform_product_category_attributes
+            (platform_id, platform_product_category_id, platform_attribute_id, complex_id,
+             name, description, group_id, group_name, value_type, is_required,
+             is_collection, max_value_count, dictionary_id, is_aspect,
+             category_dependent, is_active, source_payload, synced_at)
+         SELECT $1, $2, platform_attribute_id, complex_id,
+                name, description, group_id, group_name, value_type, is_required,
+                is_collection, max_value_count, dictionary_id, is_aspect,
+                category_dependent, true, source_payload, $4::timestamptz
+         FROM jsonb_to_recordset($3::jsonb) AS row(
+             platform_attribute_id text, complex_id text, name text, description text,
+             group_id text, group_name text, value_type text, is_required boolean,
+             is_collection boolean, max_value_count integer, dictionary_id text,
+             is_aspect boolean, category_dependent boolean, source_payload jsonb)
+         ON CONFLICT (platform_id, platform_product_category_id, platform_attribute_id, complex_id)
+         DO UPDATE SET name = EXCLUDED.name, description = EXCLUDED.description,
+             group_id = EXCLUDED.group_id, group_name = EXCLUDED.group_name,
+             value_type = EXCLUDED.value_type, is_required = EXCLUDED.is_required,
+             is_collection = EXCLUDED.is_collection, max_value_count = EXCLUDED.max_value_count,
+             dictionary_id = EXCLUDED.dictionary_id, is_aspect = EXCLUDED.is_aspect,
+             category_dependent = EXCLUDED.category_dependent, is_active = true,
+             source_payload = EXCLUDED.source_payload, synced_at = EXCLUDED.synced_at,
+             updated_at = now()`,
+        [platformId, categoryId, JSON.stringify(records), syncTime],
+    );
+    await tx.query(
+        `UPDATE atlas.platform_product_category_attributes
+         SET is_active = false, updated_at = now()
+         WHERE platform_id = $1 AND platform_product_category_id = $2
+           AND synced_at IS DISTINCT FROM $3::timestamptz`,
+        [platformId, categoryId, syncTime],
+    );
+    return { attribute_count: records.length };
+}
+
+async function persistDictionary(
+    tx: DatabaseTransaction, platformId: string, categoryId: string,
+    attributeId: string, values: OzonDictionaryValue[], isComplete: boolean,
+) {
+    const attribute = await tx.query<{ dictionary_id: string }>(
+        `SELECT dictionary_id FROM atlas.platform_product_category_attributes
+         WHERE id = $1 AND platform_id = $2 AND platform_product_category_id = $3
+           AND is_active AND dictionary_id IS NOT NULL`,
+        [attributeId, platformId, categoryId],
+    );
+    if (!attribute.rows[0]) throw new Error('字典属性已失效');
+    const byId = new Map<string, {
+        external_value_id: string; value: string; info: string | null;
+        picture: string | null; source_payload: OzonDictionaryValue;
+    }>();
+    for (const value of values) {
+        const id = String(value.id ?? '').trim();
+        const label = value.value?.trim() ?? '';
+        if (!id || !label) continue;
+        byId.set(id, {
+            external_value_id: id,
+            value: label,
+            info: typeof value.info === 'string' ? value.info : null,
+            picture: typeof value.picture === 'string' ? value.picture : null,
+            source_payload: value,
+        });
+    }
+    const records = [...byId.values()];
+    const syncTime = new Date();
+    if (records.length) {
+        await tx.query(
+            `INSERT INTO atlas.platform_catalog_dictionary_values
+                (platform_id, platform_product_category_id, attribute_id, dictionary_id,
+                 external_value_id, value, info, picture, source_payload, synced_at)
+             SELECT $1, $2, $3, $4, external_value_id, value, info, picture,
+                    source_payload, $6::timestamptz
+             FROM jsonb_to_recordset($5::jsonb) AS row(
+                 external_value_id text, value text, info text, picture text, source_payload jsonb)
+             ON CONFLICT (attribute_id, external_value_id) DO UPDATE
+             SET value = EXCLUDED.value, info = EXCLUDED.info,
+                 picture = EXCLUDED.picture, source_payload = EXCLUDED.source_payload,
+                 is_active = true, synced_at = EXCLUDED.synced_at, updated_at = now()`,
+            [platformId, categoryId, attributeId, attribute.rows[0].dictionary_id,
+                JSON.stringify(records), syncTime],
+        );
+    }
+    if (isComplete) {
+        await tx.query(
+            `UPDATE atlas.platform_catalog_dictionary_values
+             SET is_active = false, updated_at = now()
+             WHERE attribute_id = $1 AND synced_at IS DISTINCT FROM $2::timestamptz`,
+            [attributeId, syncTime],
+        );
+    }
+    return { dictionary_value_count: records.length, is_complete: isComplete };
+}
+
+async function finishRun(
+    database: Database, runId: string,
+    status: 'failed' | 'cancelled', code: string, message: string,
+) {
+    await database.transaction((tx) => updateRun(tx, runId, status, null, code, message));
+}
+
+async function updateRun(
+    tx: DatabaseTransaction, runId: string,
+    status: 'succeeded' | 'failed' | 'cancelled',
+    result: Record<string, unknown> | null,
+    code?: string, message?: string,
+) {
+    await tx.query(
+        `UPDATE atlas.platform_catalog_sync_runs
+         SET status = $2::atlas.operation_status, completed_at = now(),
+             result = $3::jsonb, error_code = $4, error_message = $5
+         WHERE id = $1 AND status IN ('queued', 'running')`,
+        [runId, status, JSON.stringify(result ?? {}), code ?? null, message ?? null],
+    );
+}