R524809
544 lines
21 KiB
TypeScript
544 lines
21 KiB
TypeScript
import type { Database } from '@atlas/database';
|
||
import type { OutboxJob } from './outbox-dispatcher.js';
|
||
import {
|
||
fetchOzonCategoryAttributes,
|
||
fetchOzonCategoryTree,
|
||
fetchOzonDictionaryValues,
|
||
OzonCredentialError,
|
||
OzonRequestError,
|
||
type OzonCategoryAttribute,
|
||
type OzonCategoryNode,
|
||
type OzonDictionaryValue,
|
||
} from './ozon-client.js';
|
||
import { decryptStoreSecret } from './store-secret.crypto.js';
|
||
import { workerConfig } from './config.js';
|
||
|
||
export type CatalogSyncScope = 'tree' | 'attributes' | 'dictionary';
|
||
type CatalogSyncJob = OutboxJob & {
|
||
payload: {
|
||
store_id: string;
|
||
secret_id: string;
|
||
operation_id: string;
|
||
platform_id: string;
|
||
scope: CatalogSyncScope;
|
||
category_id?: string;
|
||
dictionary_id?: string;
|
||
query?: string;
|
||
};
|
||
};
|
||
|
||
type SyncSource = {
|
||
platform_code: string;
|
||
platform_client_id: string;
|
||
secret_ciphertext: Buffer;
|
||
secret_nonce: Buffer;
|
||
};
|
||
type CategorySource = {
|
||
external_category_id: string;
|
||
external_type_id: string | null;
|
||
};
|
||
|
||
/** Cache is store-scoped because Ozon returns it under a seller credential. */
|
||
export async function syncOzonCatalog(database: Database, job: CatalogSyncJob) {
|
||
const source = await loadSource(database, job);
|
||
if (!source) return cancelStaleOperation(database, job);
|
||
if (source.platform_code.toUpperCase() !== 'OZON') {
|
||
return failOperation(
|
||
database,
|
||
job,
|
||
'PLATFORM_CATALOG_UNSUPPORTED',
|
||
'当前平台暂不支持类目缓存同步',
|
||
);
|
||
}
|
||
|
||
const apiKey = decryptStoreSecret(
|
||
source.secret_ciphertext,
|
||
source.secret_nonce,
|
||
workerConfig.storeSecretEncryptionKey,
|
||
);
|
||
try {
|
||
let result: Record<string, unknown>;
|
||
if (job.payload.scope === 'tree') {
|
||
const nodes = await fetchOzonCategoryTree(
|
||
source.platform_client_id,
|
||
apiKey,
|
||
);
|
||
result = await persistCategoryTree(
|
||
database,
|
||
job.workspaceId,
|
||
job.payload.store_id,
|
||
job.payload.platform_id,
|
||
nodes,
|
||
);
|
||
} else {
|
||
const category = await requireCategory(
|
||
database,
|
||
job.workspaceId,
|
||
job.payload.store_id,
|
||
job.payload.platform_id,
|
||
job.payload.category_id,
|
||
);
|
||
if (!category.external_type_id) {
|
||
return failOperation(
|
||
database,
|
||
job,
|
||
'CATEGORY_TYPE_UNAVAILABLE',
|
||
'该类目缺少 Ozon 商品类型，无法同步属性',
|
||
);
|
||
}
|
||
if (job.payload.scope === 'attributes') {
|
||
const attributes = await fetchOzonCategoryAttributes(
|
||
source.platform_client_id,
|
||
apiKey,
|
||
category.external_category_id,
|
||
category.external_type_id,
|
||
);
|
||
result = await persistAttributes(
|
||
database,
|
||
job.workspaceId,
|
||
job.payload.store_id,
|
||
job.payload.platform_id,
|
||
job.payload.category_id ?? '',
|
||
attributes,
|
||
);
|
||
} else {
|
||
const dictionary = await requireDictionaryAttribute(
|
||
database,
|
||
job.workspaceId,
|
||
job.payload.store_id,
|
||
job.payload.platform_id,
|
||
job.payload.category_id,
|
||
job.payload.dictionary_id,
|
||
);
|
||
const values = await fetchOzonDictionaryValues(
|
||
source.platform_client_id,
|
||
apiKey,
|
||
{
|
||
attributeId: dictionary.platform_attribute_id,
|
||
categoryId: category.external_category_id,
|
||
typeId: category.external_type_id,
|
||
query: job.payload.query,
|
||
},
|
||
);
|
||
result = await persistDictionaryValues(
|
||
database,
|
||
job.workspaceId,
|
||
job.payload.store_id,
|
||
job.payload.platform_id,
|
||
job.payload.category_id ?? '',
|
||
job.payload.dictionary_id ?? '',
|
||
values,
|
||
);
|
||
}
|
||
}
|
||
await succeedOperation(database, job, result);
|
||
} catch (error) {
|
||
if (error instanceof OzonCredentialError) {
|
||
await failOperation(
|
||
database,
|
||
job,
|
||
error.status === 403
|
||
? 'OZON_CATALOG_CREDENTIAL_FORBIDDEN'
|
||
: 'OZON_CATALOG_CREDENTIAL_UNAUTHORIZED',
|
||
error.status === 403
|
||
? 'Ozon 拒绝读取类目：请检查 API Key 权限'
|
||
: 'Ozon 未接受当前门店凭据',
|
||
);
|
||
return;
|
||
}
|
||
if (error instanceof OzonRequestError) {
|
||
await failOperation(
|
||
database,
|
||
job,
|
||
'OZON_CATALOG_REQUEST_REJECTED',
|
||
`Ozon 拒绝类目同步请求（HTTP ${error.status}）`,
|
||
);
|
||
return;
|
||
}
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
export async function markCatalogSyncUnavailable(
|
||
database: Database,
|
||
job: CatalogSyncJob,
|
||
_error: Error,
|
||
) {
|
||
await failOperation(
|
||
database,
|
||
job,
|
||
'OZON_CATALOG_SYNC_UNAVAILABLE',
|
||
'暂时无法同步 Ozon 类目数据，请稍后重试',
|
||
);
|
||
}
|
||
|
||
async function loadSource(database: Database, job: CatalogSyncJob) {
|
||
return database.workspaceTransaction(job.workspaceId, async (tx) => {
|
||
await tx.query(
|
||
`UPDATE atlas.operations SET status = 'running', started_at = coalesce(started_at, now()),
|
||
attempt_count = attempt_count + 1
|
||
WHERE id = $1 AND status IN ('queued', 'running')`,
|
||
[job.payload.operation_id],
|
||
);
|
||
const source = await tx.query<SyncSource>(
|
||
`SELECT platform.code AS platform_code, store.platform_client_id,
|
||
secret.secret_ciphertext, secret.secret_nonce
|
||
FROM atlas.operations operation
|
||
JOIN atlas.stores store ON store.id = $2 AND store.deleted_at IS NULL
|
||
JOIN atlas.platforms platform ON platform.id = store.platform_id AND platform.id = $3
|
||
JOIN atlas.store_secrets secret ON secret.id = $4 AND secret.store_id = store.id
|
||
WHERE operation.id = $1 AND operation.kind IN (
|
||
'store_category_tree_sync',
|
||
'store_category_attribute_sync',
|
||
'store_attribute_dictionary_sync'
|
||
)
|
||
AND store.credential_status = 'valid'
|
||
AND secret.kind = 'seller_api_key' AND secret.revoked_at IS NULL
|
||
LIMIT 1`,
|
||
[
|
||
job.payload.operation_id,
|
||
job.payload.store_id,
|
||
job.payload.platform_id,
|
||
job.payload.secret_id,
|
||
],
|
||
);
|
||
return source.rows[0] ?? null;
|
||
});
|
||
}
|
||
|
||
async function cancelStaleOperation(database: Database, job: CatalogSyncJob) {
|
||
await database.workspaceTransaction(job.workspaceId, (tx) =>
|
||
tx.query(
|
||
`UPDATE atlas.operations SET status = 'cancelled', completed_at = now(),
|
||
error_code = 'CATALOG_SYNC_SOURCE_REPLACED',
|
||
error_message = '门店凭据或发布配置已变化，旧类目同步已取消'
|
||
WHERE id = $1 AND status IN ('queued', 'running')`,
|
||
[job.payload.operation_id],
|
||
),
|
||
);
|
||
}
|
||
|
||
async function succeedOperation(
|
||
database: Database,
|
||
job: CatalogSyncJob,
|
||
result: Record<string, unknown>,
|
||
) {
|
||
await database.workspaceTransaction(job.workspaceId, (tx) =>
|
||
tx.query(
|
||
`UPDATE atlas.operations SET status = 'succeeded', completed_at = now(),
|
||
result = $2::jsonb, error_code = NULL, error_message = NULL
|
||
WHERE id = $1 AND status IN ('queued', 'running')`,
|
||
[job.payload.operation_id, JSON.stringify({ scope: job.payload.scope, ...result })],
|
||
),
|
||
);
|
||
}
|
||
|
||
async function failOperation(
|
||
database: Database,
|
||
job: CatalogSyncJob,
|
||
code: string,
|
||
message: string,
|
||
) {
|
||
await database.workspaceTransaction(job.workspaceId, (tx) =>
|
||
tx.query(
|
||
`UPDATE atlas.operations SET status = 'failed', completed_at = now(),
|
||
error_code = $2, error_message = $3
|
||
WHERE id = $1 AND status IN ('queued', 'running')`,
|
||
[job.payload.operation_id, code, message],
|
||
),
|
||
);
|
||
}
|
||
|
||
async function requireCategory(
|
||
database: Database,
|
||
workspaceId: string,
|
||
storeId: string,
|
||
platformId: string,
|
||
categoryId: string | undefined,
|
||
) {
|
||
if (!categoryId) throw new Error('Catalog sync category is missing');
|
||
const result = await database.workspaceTransaction(workspaceId, (tx) => tx.query<CategorySource>(
|
||
`SELECT external_category_id, external_type_id
|
||
FROM atlas.store_product_categories
|
||
WHERE id = $1 AND store_id = $2 AND platform_id = $3 AND is_active`,
|
||
[categoryId, storeId, platformId],
|
||
));
|
||
if (!result.rows[0]) throw new Error('Catalog sync category no longer exists');
|
||
return result.rows[0];
|
||
}
|
||
|
||
async function requireDictionaryAttribute(
|
||
database: Database,
|
||
workspaceId: string,
|
||
storeId: string,
|
||
platformId: string,
|
||
categoryId: string | undefined,
|
||
dictionaryId: string | undefined,
|
||
) {
|
||
if (!categoryId || !dictionaryId) throw new Error('Catalog dictionary input is missing');
|
||
const result = await database.workspaceTransaction(workspaceId, (tx) => tx.query<{ platform_attribute_id: string }>(
|
||
`SELECT platform_attribute_id
|
||
FROM atlas.store_product_category_attributes
|
||
WHERE store_id = $1 AND platform_id = $2 AND store_product_category_id = $3 AND dictionary_id = $4
|
||
LIMIT 1`,
|
||
[storeId, platformId, categoryId, dictionaryId],
|
||
));
|
||
if (!result.rows[0]) throw new Error('Catalog dictionary is no longer valid for this category');
|
||
return result.rows[0];
|
||
}
|
||
|
||
type FlatCategory = {
|
||
external_category_id: string;
|
||
external_type_id: string;
|
||
parent_key: string | null;
|
||
name: string;
|
||
path: string;
|
||
is_active: boolean;
|
||
source_payload: OzonCategoryNode;
|
||
};
|
||
|
||
async function persistCategoryTree(
|
||
database: Database,
|
||
workspaceId: string,
|
||
storeId: string,
|
||
platformId: string,
|
||
roots: OzonCategoryNode[],
|
||
) {
|
||
// Ozon 的树偶尔会在不同分支重复返回同一个类目；缓存主键是
|
||
// (category_id, type_id)，因此保留最后一个完整路径而不让批量 upsert 冲突。
|
||
const records = [
|
||
...new Map(
|
||
flattenCategories(roots).map((record) => [
|
||
categoryKey(record.external_category_id, record.external_type_id),
|
||
record,
|
||
]),
|
||
).values(),
|
||
];
|
||
const syncTime = new Date();
|
||
await database.workspaceTransaction(workspaceId, async (tx) => {
|
||
await tx.query(
|
||
`WITH incoming AS (
|
||
SELECT * FROM jsonb_to_recordset($3::jsonb) AS value(
|
||
external_category_id text, external_type_id text, parent_key text,
|
||
name text, path text, is_active boolean, source_payload jsonb
|
||
)
|
||
)
|
||
INSERT INTO atlas.store_product_categories
|
||
(workspace_id, store_id, platform_id, external_category_id, external_type_id, name, path,
|
||
is_active, source_payload, synced_at)
|
||
SELECT current_setting('app.workspace_id')::uuid, $1, $2, external_category_id, external_type_id, name, path,
|
||
is_active, source_payload, $4::timestamptz
|
||
FROM incoming
|
||
ON CONFLICT (store_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()`,
|
||
[storeId, platformId, JSON.stringify(records), syncTime],
|
||
);
|
||
await tx.query(
|
||
`WITH incoming AS (
|
||
SELECT * FROM jsonb_to_recordset($2::jsonb) AS value(
|
||
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.store_product_categories child
|
||
ON child.store_id = $1
|
||
AND child.external_category_id = incoming.external_category_id
|
||
AND child.external_type_id = incoming.external_type_id
|
||
LEFT JOIN atlas.store_product_categories parent
|
||
ON parent.store_id = $1
|
||
AND parent.external_category_id || ':' || parent.external_type_id = incoming.parent_key
|
||
)
|
||
UPDATE atlas.store_product_categories category
|
||
SET parent_id = resolved.parent_id, updated_at = now()
|
||
FROM resolved WHERE category.id = resolved.child_id`,
|
||
[storeId, JSON.stringify(records)],
|
||
);
|
||
await tx.query(
|
||
`UPDATE atlas.store_product_categories SET is_active = false, updated_at = now()
|
||
WHERE store_id = $1 AND synced_at IS DISTINCT FROM $2::timestamptz`,
|
||
[storeId, syncTime],
|
||
);
|
||
});
|
||
return {
|
||
category_count: records.length,
|
||
active_category_count: records.filter((record) => record.is_active).length,
|
||
};
|
||
}
|
||
|
||
export function flattenCategories(
|
||
nodes: OzonCategoryNode[],
|
||
path: string[] = [],
|
||
parentKey: string | null = null,
|
||
inheritedCategory?: { id: string; name: string },
|
||
): FlatCategory[] {
|
||
const flattened: FlatCategory[] = [];
|
||
for (const node of nodes) {
|
||
// Ozon's last level may contain only type_name/type_id. It is still a
|
||
// publishable option for the closest description category, so inherit
|
||
// its category ID while retaining the type's own display name and ID.
|
||
const explicitCategoryId = String(node.description_category_id ?? '').trim();
|
||
const explicitCategoryName = node.category_name?.trim() ?? '';
|
||
const categoryId = explicitCategoryId || inheritedCategory?.id || '';
|
||
const categoryName = explicitCategoryName || inheritedCategory?.name || '';
|
||
const typeId = String(node.type_id ?? 0);
|
||
const typeName = node.type_name?.trim() ?? '';
|
||
const name = explicitCategoryName || typeName || categoryName;
|
||
if (!categoryId || !name) continue;
|
||
const nextPath = [...path, name];
|
||
const key = categoryKey(categoryId, typeId);
|
||
flattened.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,
|
||
});
|
||
const nextInheritedCategory = explicitCategoryId && explicitCategoryName
|
||
? { id: explicitCategoryId, name: explicitCategoryName }
|
||
: inheritedCategory;
|
||
flattened.push(
|
||
...flattenCategories(
|
||
node.children ?? [],
|
||
nextPath,
|
||
key,
|
||
nextInheritedCategory,
|
||
),
|
||
);
|
||
}
|
||
return flattened;
|
||
}
|
||
|
||
function categoryKey(categoryId: string, typeId: string) {
|
||
// Ozon category/type IDs are scalar identifiers; a printable separator is
|
||
// intentional because PostgreSQL text/jsonb forbids embedded NUL bytes.
|
||
return `${categoryId}:${typeId}`;
|
||
}
|
||
|
||
type CachedAttribute = {
|
||
platform_attribute_id: string;
|
||
complex_id: string;
|
||
name: string;
|
||
value_type: 'text' | 'integer' | 'decimal' | 'boolean' | 'dictionary';
|
||
is_required: boolean;
|
||
is_collection: boolean;
|
||
max_value_count: number;
|
||
dictionary_id: string | null;
|
||
is_aspect: boolean;
|
||
source_payload: OzonCategoryAttribute;
|
||
};
|
||
|
||
async function persistAttributes(
|
||
database: Database,
|
||
workspaceId: string,
|
||
storeId: string,
|
||
platformId: string,
|
||
categoryId: string,
|
||
attributes: OzonCategoryAttribute[],
|
||
) {
|
||
const records = attributes
|
||
.map(toCachedAttribute)
|
||
.filter((attribute): attribute is CachedAttribute => attribute !== null);
|
||
await database.workspaceTransaction(workspaceId, async (tx) => {
|
||
await tx.query(
|
||
`DELETE FROM atlas.store_product_category_attributes
|
||
WHERE store_id = $1 AND store_product_category_id = $2`,
|
||
[storeId, categoryId],
|
||
);
|
||
if (!records.length) return;
|
||
await tx.query(
|
||
`INSERT INTO atlas.store_product_category_attributes
|
||
(workspace_id, store_id, platform_id, store_product_category_id, platform_attribute_id, complex_id,
|
||
name, value_type, is_required, is_collection, max_value_count,
|
||
dictionary_id, is_aspect, source_payload, synced_at)
|
||
SELECT current_setting('app.workspace_id')::uuid, $1, $2, $3, platform_attribute_id, complex_id, name, value_type,
|
||
is_required, is_collection, max_value_count, dictionary_id,
|
||
is_aspect, source_payload, now()
|
||
FROM jsonb_to_recordset($4::jsonb) AS value(
|
||
platform_attribute_id text, complex_id text, name text, value_type text,
|
||
is_required boolean, is_collection boolean, max_value_count integer,
|
||
dictionary_id text, is_aspect boolean, source_payload jsonb
|
||
)`,
|
||
[storeId, platformId, categoryId, JSON.stringify(records)],
|
||
);
|
||
});
|
||
return { attribute_count: records.length };
|
||
}
|
||
|
||
function toCachedAttribute(value: OzonCategoryAttribute): CachedAttribute | null {
|
||
const id = String(value.id ?? '').trim();
|
||
const name = value.name?.trim();
|
||
if (!id || !name) return null;
|
||
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));
|
||
return {
|
||
platform_attribute_id: id,
|
||
complex_id: String(value.complex_id ?? 0),
|
||
name,
|
||
value_type: hasDictionary ? 'dictionary' : mapAttributeType(value.type),
|
||
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,
|
||
source_payload: value,
|
||
};
|
||
}
|
||
|
||
function mapAttributeType(value: string | null | undefined): CachedAttribute['value_type'] {
|
||
switch (value?.toLowerCase()) {
|
||
case 'integer':
|
||
return 'integer';
|
||
case 'decimal':
|
||
case 'number':
|
||
return 'decimal';
|
||
case 'boolean':
|
||
return 'boolean';
|
||
default:
|
||
return 'text';
|
||
}
|
||
}
|
||
|
||
async function persistDictionaryValues(
|
||
database: Database,
|
||
workspaceId: string,
|
||
storeId: string,
|
||
platformId: string,
|
||
categoryId: string,
|
||
dictionaryId: string,
|
||
values: OzonDictionaryValue[],
|
||
) {
|
||
const records = values
|
||
.map((value) => ({
|
||
external_value_id: String(value.id ?? '').trim(),
|
||
value: value.value?.trim(),
|
||
source_payload: value,
|
||
}))
|
||
.filter((value): value is { external_value_id: string; value: string; source_payload: OzonDictionaryValue } =>
|
||
!!value.external_value_id && !!value.value,
|
||
);
|
||
if (!records.length) return { dictionary_value_count: 0 };
|
||
await database.workspaceTransaction(workspaceId, (tx) => tx.query(
|
||
`INSERT INTO atlas.store_attribute_dictionary_values
|
||
(workspace_id, store_id, platform_id, store_product_category_id, dictionary_id, external_value_id, value, is_active, source_payload, synced_at)
|
||
SELECT current_setting('app.workspace_id')::uuid, $1, $2, $3, $4, external_value_id, value, true, source_payload, now()
|
||
FROM jsonb_to_recordset($5::jsonb) AS value(
|
||
external_value_id text, value text, source_payload jsonb
|
||
)
|
||
ON CONFLICT (store_id, store_product_category_id, dictionary_id, external_value_id) DO UPDATE
|
||
SET value = EXCLUDED.value, is_active = true, source_payload = EXCLUDED.source_payload,
|
||
synced_at = EXCLUDED.synced_at, updated_at = now()`,
|
||
[storeId, platformId, categoryId, dictionaryId, JSON.stringify(records)],
|
||
));
|
||
return { dictionary_value_count: records.length };
|
||
}
|