R524809
509 lines
19 KiB
TypeScript
509 lines
19 KiB
TypeScript
import type {
|
|
ProductAttributeSchema,
|
|
ProductCatalogCacheMetadata,
|
|
ProductCatalogRefreshRun,
|
|
ProductEditorValidation,
|
|
ProductPublishPreflight,
|
|
} from '@atlas/contracts';
|
|
import type { Database, DatabaseTransaction } from '@atlas/database';
|
|
import { ApiError } from '../common/api-error.js';
|
|
import {
|
|
enqueuePlatformCatalogRun,
|
|
type PlatformCatalogRunRow,
|
|
} from '../platform-catalog/platform-catalog-run.js';
|
|
import {
|
|
productCatalogCacheMetadata as metadata,
|
|
productCatalogResourceStatus as statusOf,
|
|
productCatalogStateWithRun as withRun,
|
|
type ProductCatalogResourceState as ResourceState,
|
|
} from './product-editor.catalog-cache.js';
|
|
import type { ProfileRow } from './product-editor.models.js';
|
|
|
|
type PublishPreflightAccess = {
|
|
requireProfileForPlatform: (
|
|
transaction: DatabaseTransaction,
|
|
profileId: string,
|
|
platformId: string,
|
|
) => Promise<ProfileRow>;
|
|
requireActiveCategory: (
|
|
transaction: DatabaseTransaction,
|
|
platformId: string,
|
|
categoryId: string,
|
|
) => Promise<unknown>;
|
|
attributeSchemaForCategory: (
|
|
transaction: DatabaseTransaction,
|
|
platformId: string,
|
|
categoryId: string,
|
|
) => Promise<ProductAttributeSchema[]>;
|
|
validation: (
|
|
workspaceId: string,
|
|
profileId: string,
|
|
platformId: string,
|
|
variantIds: string[],
|
|
) => Promise<ProductEditorValidation>;
|
|
};
|
|
|
|
type PreparedCatalog = {
|
|
schema: ProductCatalogCacheMetadata;
|
|
dictionaries: ProductPublishPreflight['catalog']['dictionaries'];
|
|
runs: PlatformCatalogRunRow[];
|
|
blockingRunIds: string[];
|
|
hasBlockingResources: boolean;
|
|
};
|
|
|
|
type UsedDictionary = {
|
|
attribute_id: string;
|
|
platform_attribute_id: string;
|
|
name: string;
|
|
};
|
|
|
|
/**
|
|
* 首次发布预检只刷新当前商品实际依赖的平台目录资源。
|
|
* API 只创建平台任务并观察 Worker 结果，绝不在请求进程直接调用 Ozon。
|
|
*/
|
|
export class ProductEditorPublishPreflightOperations {
|
|
public constructor(
|
|
private readonly database: Database,
|
|
private readonly access: PublishPreflightAccess,
|
|
) {}
|
|
|
|
async preflight(
|
|
workspaceId: string,
|
|
userId: string,
|
|
profileId: string,
|
|
platformId: string,
|
|
variantIds: string[],
|
|
): Promise<{ status: number; body: { data: ProductPublishPreflight } }> {
|
|
const deadline = Date.now() + 5_000;
|
|
const observedRuns = new Map<string, PlatformCatalogRunRow>();
|
|
let prepared = await this.prepare(
|
|
workspaceId,
|
|
userId,
|
|
profileId,
|
|
platformId,
|
|
variantIds,
|
|
true,
|
|
);
|
|
rememberRuns(observedRuns, prepared.runs);
|
|
|
|
// 第一轮通常等待 schema；成功后第二轮才知道本商品实际引用哪些字典。
|
|
for (
|
|
let pass = 0;
|
|
pass < 2 && prepared.blockingRunIds.length;
|
|
pass += 1
|
|
) {
|
|
const succeeded = await this.waitForRuns(
|
|
prepared.blockingRunIds,
|
|
deadline,
|
|
);
|
|
prepared = await this.prepare(
|
|
workspaceId,
|
|
userId,
|
|
profileId,
|
|
platformId,
|
|
variantIds,
|
|
succeeded && Date.now() < deadline,
|
|
);
|
|
rememberRuns(observedRuns, prepared.runs);
|
|
if (!succeeded) break;
|
|
}
|
|
|
|
const validation = await this.access.validation(
|
|
workspaceId,
|
|
profileId,
|
|
platformId,
|
|
variantIds,
|
|
);
|
|
const currentRuns = await this.currentRuns([...observedRuns.keys()]);
|
|
const refreshing = prepared.hasBlockingResources;
|
|
const result: ProductPublishPreflight = {
|
|
status: refreshing
|
|
? 'refreshing'
|
|
: validation.ready
|
|
? 'ready'
|
|
: 'blocked',
|
|
validation,
|
|
catalog: {
|
|
schema: prepared.schema,
|
|
dictionaries: prepared.dictionaries,
|
|
},
|
|
refresh_runs: currentRuns.map(toRefreshRun),
|
|
};
|
|
return {
|
|
status: refreshing ? 202 : 200,
|
|
body: { data: result },
|
|
};
|
|
}
|
|
|
|
private async prepare(
|
|
workspaceId: string,
|
|
userId: string,
|
|
profileId: string,
|
|
platformId: string,
|
|
variantIds: string[],
|
|
enqueue: boolean,
|
|
): Promise<PreparedCatalog> {
|
|
return this.database.workspaceTransaction(
|
|
workspaceId,
|
|
async (transaction) => {
|
|
const profile = await this.access.requireProfileForPlatform(
|
|
transaction,
|
|
profileId,
|
|
platformId,
|
|
);
|
|
if (!profile.platform_product_category_id)
|
|
throw new ApiError(
|
|
422,
|
|
'CATEGORY_REQUIRED',
|
|
'首次上架前请先选择产品类目',
|
|
);
|
|
await this.access.requireActiveCategory(
|
|
transaction,
|
|
platformId,
|
|
profile.platform_product_category_id,
|
|
);
|
|
await this.requirePublishableVariants(
|
|
transaction,
|
|
profile,
|
|
variantIds,
|
|
);
|
|
|
|
let sourceRevision: number | null = null;
|
|
const runs: PlatformCatalogRunRow[] = [];
|
|
const blockingRunIds: string[] = [];
|
|
const enqueueResource = async (input: {
|
|
scope: 'attributes' | 'dictionary';
|
|
scopeKey: string;
|
|
categoryId: string;
|
|
attributeId?: string;
|
|
priority: 'high' | 'low';
|
|
}) => {
|
|
if (sourceRevision === null) {
|
|
const source = await transaction.query<{
|
|
revision: number;
|
|
}>(
|
|
`SELECT revision FROM atlas.platform_catalog_sources
|
|
WHERE platform_id = $1 FOR UPDATE`,
|
|
[platformId],
|
|
);
|
|
if (!source.rows[0])
|
|
throw new ApiError(
|
|
422,
|
|
'CATALOG_SOURCE_REQUIRED',
|
|
'平台目录同步来源尚未配置',
|
|
);
|
|
sourceRevision = source.rows[0].revision;
|
|
}
|
|
const run = await enqueuePlatformCatalogRun(transaction, {
|
|
platformId,
|
|
scope: input.scope,
|
|
scopeKey: input.scopeKey,
|
|
categoryId: input.categoryId,
|
|
attributeId: input.attributeId,
|
|
sourceRevision,
|
|
trigger: 'publish',
|
|
priority: input.priority,
|
|
userId,
|
|
});
|
|
runs.push(run);
|
|
return run;
|
|
};
|
|
|
|
const categoryId = profile.platform_product_category_id;
|
|
let schemaState = await this.resourceState(
|
|
transaction,
|
|
platformId,
|
|
'attributes',
|
|
`attributes:${categoryId}`,
|
|
);
|
|
const schemaStatus = statusOf(schemaState);
|
|
if (enqueue && schemaStatus !== 'fresh') {
|
|
const run = await enqueueResource({
|
|
scope: 'attributes',
|
|
scopeKey: `attributes:${categoryId}`,
|
|
categoryId,
|
|
priority: schemaStatus === 'stale' ? 'low' : 'high',
|
|
});
|
|
schemaState = withRun(schemaState, run);
|
|
if (
|
|
schemaStatus === 'expired' ||
|
|
schemaStatus === 'missing'
|
|
)
|
|
blockingRunIds.push(run.id);
|
|
} else if (
|
|
schemaStatus === 'expired' ||
|
|
schemaStatus === 'missing'
|
|
) {
|
|
if (schemaState.refresh_run_id)
|
|
blockingRunIds.push(schemaState.refresh_run_id);
|
|
}
|
|
|
|
const schemaUsable =
|
|
schemaStatus === 'fresh' || schemaStatus === 'stale';
|
|
const dictionaries: ProductPublishPreflight['catalog']['dictionaries'] =
|
|
[];
|
|
if (schemaUsable) {
|
|
const schema = await this.access.attributeSchemaForCategory(
|
|
transaction,
|
|
platformId,
|
|
categoryId,
|
|
);
|
|
const used = await this.usedDictionaries(
|
|
transaction,
|
|
profileId,
|
|
variantIds,
|
|
schema,
|
|
);
|
|
for (const dictionary of used) {
|
|
let state = await this.resourceState(
|
|
transaction,
|
|
platformId,
|
|
'dictionary',
|
|
`dictionary:${dictionary.attribute_id}`,
|
|
);
|
|
const status = statusOf(state);
|
|
if (enqueue && status !== 'fresh') {
|
|
const run = await enqueueResource({
|
|
scope: 'dictionary',
|
|
scopeKey: `dictionary:${dictionary.attribute_id}`,
|
|
categoryId,
|
|
attributeId: dictionary.attribute_id,
|
|
priority: status === 'stale' ? 'low' : 'high',
|
|
});
|
|
state = withRun(state, run);
|
|
if (status === 'expired' || status === 'missing')
|
|
blockingRunIds.push(run.id);
|
|
} else if (
|
|
status === 'expired' ||
|
|
status === 'missing'
|
|
) {
|
|
if (state.refresh_run_id)
|
|
blockingRunIds.push(state.refresh_run_id);
|
|
}
|
|
dictionaries.push({
|
|
...dictionary,
|
|
cache: metadata(state, status),
|
|
});
|
|
}
|
|
}
|
|
|
|
return {
|
|
schema: metadata(schemaState, schemaStatus),
|
|
dictionaries,
|
|
runs,
|
|
blockingRunIds: [...new Set(blockingRunIds)],
|
|
hasBlockingResources:
|
|
!schemaUsable ||
|
|
dictionaries.some(
|
|
(item) =>
|
|
item.cache.status === 'expired' ||
|
|
item.cache.status === 'missing',
|
|
),
|
|
};
|
|
},
|
|
);
|
|
}
|
|
|
|
/** 发布入口只能选择当前 profile 下尚未首次导入的活动 SKU。 */
|
|
private async requirePublishableVariants(
|
|
transaction: DatabaseTransaction,
|
|
profile: ProfileRow,
|
|
variantIds: string[],
|
|
) {
|
|
const result = await transaction.query<{ id: string; status: string }>(
|
|
`SELECT variant.id, listing.status
|
|
FROM atlas.product_variants variant
|
|
JOIN atlas.product_variant_platform_listings listing
|
|
ON listing.product_variant_id = variant.id
|
|
AND listing.profile_id = $1
|
|
WHERE variant.product_id = $2 AND variant.deleted_at IS NULL
|
|
AND variant.id = ANY($3::uuid[])`,
|
|
[profile.id, profile.product_id, variantIds],
|
|
);
|
|
if (result.rows.length !== new Set(variantIds).size)
|
|
throw new ApiError(
|
|
422,
|
|
'PUBLISH_VARIANTS_INVALID',
|
|
'待发布 SKU 不属于当前商品，或已经被删除',
|
|
);
|
|
if (result.rows.some((item) => item.status === 'submitting'))
|
|
throw new ApiError(
|
|
409,
|
|
'SKU_IMPORT_IN_PROGRESS',
|
|
'所选 SKU 正在首次上架，请等待当前任务完成',
|
|
);
|
|
if (result.rows.some((item) => item.status === 'imported'))
|
|
throw new ApiError(
|
|
422,
|
|
'SKU_ALREADY_IMPORTED',
|
|
'所选 SKU 已经首次导入，不能再次上架',
|
|
);
|
|
if (
|
|
result.rows.some(
|
|
(item) => !['not_listed', 'failed'].includes(item.status),
|
|
)
|
|
)
|
|
throw new ApiError(
|
|
422,
|
|
'SKU_NOT_PUBLISHABLE',
|
|
'所选 SKU 当前状态不允许首次上架',
|
|
);
|
|
}
|
|
|
|
/** 从已确认草稿中提取本次 SKU 集合真正使用的字典属性。 */
|
|
private async usedDictionaries(
|
|
transaction: DatabaseTransaction,
|
|
profileId: string,
|
|
variantIds: string[],
|
|
schema: ProductAttributeSchema[],
|
|
): Promise<UsedDictionary[]> {
|
|
const definitions = new Map(
|
|
schema
|
|
.filter(
|
|
(item) =>
|
|
item.value_type === 'dictionary' &&
|
|
item.dictionary_id &&
|
|
!item.editor_field,
|
|
)
|
|
.map((item) => [
|
|
`${item.platform_attribute_id}\u0000${item.complex_id}`,
|
|
item,
|
|
]),
|
|
);
|
|
const attributes = await transaction.query<{
|
|
platform_attribute_id: string;
|
|
complex_id: string;
|
|
values: unknown;
|
|
}>(
|
|
`SELECT platform_attribute_id, complex_id, values
|
|
FROM atlas.product_attributes
|
|
WHERE profile_id = $1 AND confirmation_status = 'confirmed'
|
|
AND (scope = 'common' OR product_variant_id = ANY($2::uuid[]))`,
|
|
[profileId, variantIds],
|
|
);
|
|
const used = new Map<string, UsedDictionary>();
|
|
// 内建品牌等字典字段也需要平台候选值，不能绕过发布目录有效期检查。
|
|
for (const definition of schema.filter(item => item.editor_field && item.value_type === 'dictionary' && item.dictionary_id)) {
|
|
used.set(definition.attribute_id, {attribute_id: definition.attribute_id, platform_attribute_id: definition.platform_attribute_id, name: definition.name});
|
|
}
|
|
for (const attribute of attributes.rows) {
|
|
if (!hasStringValue(attribute.values)) continue;
|
|
const definition = definitions.get(
|
|
`${attribute.platform_attribute_id}\u0000${attribute.complex_id}`,
|
|
);
|
|
if (!definition || used.has(definition.attribute_id)) continue;
|
|
used.set(definition.attribute_id, {
|
|
attribute_id: definition.attribute_id,
|
|
platform_attribute_id: definition.platform_attribute_id,
|
|
name: definition.name,
|
|
});
|
|
}
|
|
return [...used.values()];
|
|
}
|
|
|
|
private async resourceState(
|
|
transaction: DatabaseTransaction,
|
|
platformId: string,
|
|
scope: 'attributes' | 'dictionary',
|
|
scopeKey: string,
|
|
): Promise<ResourceState> {
|
|
const result = await transaction.query<ResourceState>(
|
|
`SELECT state.last_success_at,
|
|
coalesce(state.is_complete, false) AS is_complete,
|
|
CASE WHEN $2 = 'attributes' THEN policy.attribute_fresh_days
|
|
ELSE policy.dictionary_fresh_days END AS fresh_days,
|
|
CASE WHEN $2 = 'attributes' THEN policy.attribute_expire_days
|
|
ELSE policy.dictionary_expire_days END AS expire_days,
|
|
run.id AS refresh_run_id, run.status AS refresh_status
|
|
FROM atlas.platform_catalog_refresh_policies policy
|
|
LEFT JOIN atlas.platform_catalog_resource_states state
|
|
ON state.platform_id = policy.platform_id AND state.scope_key = $3
|
|
LEFT JOIN LATERAL (
|
|
SELECT id, status
|
|
FROM atlas.platform_catalog_sync_runs
|
|
WHERE platform_id = $1 AND scope_key = $3
|
|
AND status IN ('queued', 'running')
|
|
ORDER BY created_at DESC LIMIT 1
|
|
) run ON true
|
|
WHERE policy.platform_id = $1`,
|
|
[platformId, scope, scopeKey],
|
|
);
|
|
if (!result.rows[0])
|
|
throw new ApiError(
|
|
422,
|
|
'CATALOG_POLICY_REQUIRED',
|
|
'平台目录刷新策略尚未初始化',
|
|
);
|
|
return result.rows[0];
|
|
}
|
|
|
|
/** 多个高优先级资源共用同一个五秒窗口，避免字典数量线性放大请求时长。 */
|
|
private async waitForRuns(runIds: string[], deadline: number) {
|
|
while (Date.now() < deadline) {
|
|
const result = await this.database.query<{
|
|
id: string;
|
|
status: string;
|
|
}>(
|
|
`SELECT id, status FROM atlas.platform_catalog_sync_runs
|
|
WHERE id = ANY($1::uuid[])`,
|
|
[runIds],
|
|
);
|
|
const statusById = new Map(
|
|
result.rows.map((item) => [item.id, item.status]),
|
|
);
|
|
const statuses = runIds.map(
|
|
(id) => statusById.get(id) ?? 'cancelled',
|
|
);
|
|
if (statuses.every((status) => status === 'succeeded')) return true;
|
|
if (
|
|
statuses.some((status) =>
|
|
['failed', 'cancelled'].includes(status),
|
|
)
|
|
)
|
|
return false;
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 250));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/** 等待结束后重新读取任务状态，避免响应继续展示排队时的旧快照。 */
|
|
private async currentRuns(runIds: string[]) {
|
|
if (!runIds.length) return [];
|
|
const result = await this.database.query<PlatformCatalogRunRow>(
|
|
`SELECT id, platform_id, scope, scope_key, status, trigger_kind,
|
|
priority, created_at
|
|
FROM atlas.platform_catalog_sync_runs
|
|
WHERE id = ANY($1::uuid[])
|
|
ORDER BY created_at, id`,
|
|
[runIds],
|
|
);
|
|
return result.rows;
|
|
}
|
|
}
|
|
|
|
function rememberRuns(
|
|
target: Map<string, PlatformCatalogRunRow>,
|
|
runs: PlatformCatalogRunRow[],
|
|
) {
|
|
for (const run of runs) target.set(run.id, run);
|
|
}
|
|
|
|
function toRefreshRun(run: PlatformCatalogRunRow): ProductCatalogRefreshRun {
|
|
return {
|
|
id: run.id,
|
|
platform_id: run.platform_id,
|
|
scope: run.scope,
|
|
status: run.status,
|
|
trigger_kind: run.trigger_kind,
|
|
priority: run.priority,
|
|
created_at: new Date(run.created_at).toISOString(),
|
|
};
|
|
}
|
|
|
|
function hasStringValue(value: unknown) {
|
|
return (
|
|
Array.isArray(value) &&
|
|
value.some((item) => typeof item === 'string' && item.trim())
|
|
);
|
|
}
|