R524809
489 lines
21 KiB
TypeScript
489 lines
21 KiB
TypeScript
import { Inject, Injectable } from '@nestjs/common';
|
|
import type { CollectedProductPayload } from '@atlas/contracts';
|
|
import type { Database } from '@atlas/database';
|
|
import { ApiError } from '../common/api-error.js';
|
|
import { IdempotencyService } from '../common/idempotency.service.js';
|
|
import { DATABASE } from '../database/database.module.js';
|
|
|
|
@Injectable()
|
|
export class ExtensionService {
|
|
public constructor(
|
|
@Inject(DATABASE) private readonly database: Database,
|
|
@Inject(IdempotencyService)
|
|
private readonly idempotency: IdempotencyService,
|
|
) {}
|
|
|
|
public async ingest(
|
|
workspaceId: string,
|
|
userId: string,
|
|
key: string,
|
|
payload: CollectedProductPayload,
|
|
) {
|
|
return this.idempotency.run(
|
|
{
|
|
workspaceId,
|
|
userId,
|
|
method: 'POST',
|
|
path: '/api/extension/materials',
|
|
key,
|
|
request: payload,
|
|
},
|
|
async (tx) => {
|
|
const platform = await tx.query<{ id: string; code: string }>(
|
|
`SELECT id, code FROM atlas.platforms
|
|
WHERE code = $1 AND status = 'active' AND coalesce((capabilities->>'can_be_source')::boolean, false)`,
|
|
[payload.source.platform.toUpperCase()],
|
|
);
|
|
if (!platform.rows[0])
|
|
throw new ApiError(
|
|
422,
|
|
'SOURCE_PLATFORM_UNAVAILABLE',
|
|
'该来源平台当前不可采集',
|
|
);
|
|
const existing = await this.findExistingSourceItem(
|
|
tx,
|
|
workspaceId,
|
|
platform.rows[0].id,
|
|
payload,
|
|
);
|
|
if (existing) {
|
|
await this.replaceSourceSnapshot(
|
|
tx,
|
|
workspaceId,
|
|
existing.id,
|
|
existing.product_id,
|
|
payload,
|
|
);
|
|
return {
|
|
status: 200,
|
|
body: {
|
|
data: await this.materialResponse(
|
|
tx,
|
|
existing.id,
|
|
existing.product_id,
|
|
true,
|
|
true,
|
|
),
|
|
},
|
|
};
|
|
}
|
|
const { status, sourcePrice } = sourceState(payload);
|
|
const item = await tx.query<{ id: string }>(
|
|
`INSERT INTO atlas.collection_items
|
|
(workspace_id, source_kind, source_platform_id, source_item_key, source_url, source_title, source_currency_code, source_min_price, source_max_price, status, raw_evidence, normalized_payload, parser_version, collected_at)
|
|
VALUES (current_setting('app.workspace_id')::uuid, 'platform', $1, $2, $3, $4, $5, $6, $6, $7::atlas.collection_item_status, $8, $9, $10, $11) RETURNING id`,
|
|
[
|
|
platform.rows[0].id,
|
|
payload.source.itemKey,
|
|
payload.source.canonicalUrl,
|
|
payload.common.title?.value ?? null,
|
|
sourcePrice?.currencyCode ?? null,
|
|
sourcePrice?.amount ?? null,
|
|
status,
|
|
JSON.stringify(payload.evidence),
|
|
JSON.stringify(payload),
|
|
payload.source.parserVersion,
|
|
payload.source.collectedAt,
|
|
],
|
|
);
|
|
const product = await tx.query<{ id: string }>(
|
|
`INSERT INTO atlas.products (workspace_id, collection_item_id, source_title)
|
|
VALUES (current_setting('app.workspace_id')::uuid, $1, $2) RETURNING id`,
|
|
[item.rows[0].id, payload.common.title?.value ?? null],
|
|
);
|
|
await this.writeSourceSnapshotRows(
|
|
tx,
|
|
workspaceId,
|
|
item.rows[0].id,
|
|
product.rows[0].id,
|
|
1,
|
|
payload,
|
|
false,
|
|
);
|
|
return {
|
|
status: 201,
|
|
body: {
|
|
data: await this.materialResponse(
|
|
tx,
|
|
item.rows[0].id,
|
|
product.rows[0].id,
|
|
false,
|
|
false,
|
|
),
|
|
},
|
|
};
|
|
},
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Ozon has no public, stable SPU ID. The entered URL SKU remains the source
|
|
* identity for audit purposes, while a re-collection from any sibling URL
|
|
* reuses the same item when its real source SKU set intersects this one.
|
|
*/
|
|
private async findExistingSourceItem(
|
|
tx: { query: Database['query'] },
|
|
workspaceId: string,
|
|
platformId: string,
|
|
payload: CollectedProductPayload,
|
|
): Promise<{ id: string; product_id: string } | null> {
|
|
const bySourceKey = await tx.query<{ id: string; product_id: string }>(
|
|
`SELECT c.id, p.id AS product_id
|
|
FROM atlas.collection_items c
|
|
JOIN atlas.products p ON p.collection_item_id = c.id
|
|
WHERE c.source_platform_id = $1
|
|
AND c.source_item_key = $2
|
|
AND c.deleted_at IS NULL`,
|
|
[platformId, payload.source.itemKey],
|
|
);
|
|
if (bySourceKey.rows[0]) return bySourceKey.rows[0];
|
|
if (payload.source.platform !== 'ozon') return null;
|
|
|
|
const sourceSkus = [
|
|
...new Set(
|
|
payload.variants
|
|
.map(
|
|
(variant) =>
|
|
variant.sourceSku ?? variant.sourceVariantKey,
|
|
)
|
|
.filter((sku) => sku.trim().length > 0),
|
|
),
|
|
];
|
|
if (!sourceSkus.length) return null;
|
|
// Different sibling URLs have different source_item_key values. Locking
|
|
// each real SKU in sorted order makes two concurrent first collections
|
|
// of the same public variant group observe one another's snapshot.
|
|
for (const sourceSku of sourceSkus.sort()) {
|
|
await tx.query(
|
|
`SELECT pg_advisory_xact_lock(hashtextextended($1::text, 0))`,
|
|
[`${workspaceId}:${platformId}:${sourceSku}`],
|
|
);
|
|
}
|
|
const byVariantOverlap = await tx.query<{
|
|
id: string;
|
|
product_id: string;
|
|
}>(
|
|
`SELECT DISTINCT c.id, p.id AS product_id
|
|
FROM atlas.collection_items c
|
|
JOIN atlas.products p ON p.collection_item_id = c.id
|
|
JOIN atlas.product_variants v ON v.product_id = p.id
|
|
WHERE c.source_platform_id = $1
|
|
AND c.deleted_at IS NULL
|
|
AND v.deleted_at IS NULL
|
|
AND coalesce(v.source_sku, v.source_variant_key) = ANY($2::text[])
|
|
LIMIT 2`,
|
|
[platformId, sourceSkus],
|
|
);
|
|
if (byVariantOverlap.rows.length > 1) {
|
|
throw new ApiError(
|
|
409,
|
|
'SOURCE_VARIANT_GROUP_AMBIGUOUS',
|
|
'发现多条历史采集记录包含相同 Ozon SKU，请先在采集箱处理重复记录',
|
|
);
|
|
}
|
|
return byVariantOverlap.rows[0] ?? null;
|
|
}
|
|
|
|
private async replaceSourceSnapshot(
|
|
tx: { query: Database['query'] },
|
|
workspaceId: string,
|
|
collectionItemId: string,
|
|
productId: string,
|
|
payload: CollectedProductPayload,
|
|
) {
|
|
const { status, sourcePrice } = sourceState(payload);
|
|
// Once a platform profile exists, product_variants are no longer a
|
|
// disposable collection projection: they may carry user edits and are
|
|
// referenced by platform listings. A re-collection refreshes the
|
|
// source evidence/candidates, but must not overwrite that draft or
|
|
// sever a SKU's publishing history.
|
|
const profileCount = await tx.query<{ count: string }>(
|
|
`SELECT count(*)::text AS count
|
|
FROM atlas.product_platform_profiles
|
|
WHERE product_id = $1`,
|
|
[productId],
|
|
);
|
|
const preserveEditorDraft =
|
|
Number(profileCount.rows[0]?.count ?? 0) > 0;
|
|
// 不删除旧候选/任务记录：运行中的 Worker 可能已拿到它。取消并保留审计记录，
|
|
// 新快照在查询和 Worker 中均按 source_snapshot_revision 隔离。
|
|
await tx.query(
|
|
`UPDATE atlas.collection_asset_transfers SET status = 'cancelled', completed_at = now(), updated_at = now()
|
|
WHERE collection_item_id = $1 AND status IN ('pending_selection', 'queued', 'running')`,
|
|
[collectionItemId],
|
|
);
|
|
const item = await tx.query<{
|
|
source_snapshot_revision: string | number;
|
|
}>(
|
|
`UPDATE atlas.collection_items
|
|
SET source_url = $1, source_title = $2, source_currency_code = $3, source_min_price = $4, source_max_price = $4,
|
|
status = $5::atlas.collection_item_status, raw_evidence = $6, normalized_payload = $7, parser_version = $8,
|
|
collected_at = $9, source_snapshot_revision = source_snapshot_revision + 1, revision = revision + 1, updated_at = now()
|
|
WHERE id = $10 AND deleted_at IS NULL RETURNING source_snapshot_revision`,
|
|
[
|
|
payload.source.canonicalUrl,
|
|
payload.common.title?.value ?? null,
|
|
sourcePrice?.currencyCode ?? null,
|
|
sourcePrice?.amount ?? null,
|
|
status,
|
|
JSON.stringify(payload.evidence),
|
|
JSON.stringify(payload),
|
|
payload.source.parserVersion,
|
|
payload.source.collectedAt,
|
|
collectionItemId,
|
|
],
|
|
);
|
|
const snapshotRevision = Number(item.rows[0]?.source_snapshot_revision);
|
|
if (!Number.isSafeInteger(snapshotRevision) || snapshotRevision < 1)
|
|
throw new Error('采集来源快照版本更新失败');
|
|
if (!preserveEditorDraft) {
|
|
await tx.query(
|
|
`UPDATE atlas.products
|
|
SET source_title = $1, updated_at = now(), revision = revision + 1
|
|
WHERE id = $2`,
|
|
[payload.common.title?.value ?? null, productId],
|
|
);
|
|
}
|
|
await this.writeSourceSnapshotRows(
|
|
tx,
|
|
workspaceId,
|
|
collectionItemId,
|
|
productId,
|
|
snapshotRevision,
|
|
payload,
|
|
preserveEditorDraft,
|
|
);
|
|
return snapshotRevision;
|
|
}
|
|
|
|
private async writeSourceSnapshotRows(
|
|
tx: { query: Database['query'] },
|
|
workspaceId: string,
|
|
collectionItemId: string,
|
|
productId: string,
|
|
snapshotRevision: number,
|
|
payload: CollectedProductPayload,
|
|
preserveEditorDraft: boolean,
|
|
) {
|
|
const verifiedVariants = payload.variants.filter(
|
|
(variant) => variant.sourceVariantKey.trim().length > 0,
|
|
);
|
|
const variantRows = verifiedVariants.length
|
|
? verifiedVariants
|
|
: [
|
|
{
|
|
sourceVariantKey: null,
|
|
sourceSku: undefined,
|
|
displayName: '默认 SKU',
|
|
},
|
|
];
|
|
const sourceKeys = variantRows
|
|
.map((variant) => variant.sourceVariantKey)
|
|
.filter((key): key is string => !!key);
|
|
const existing = await tx.query<{
|
|
id: string;
|
|
source_variant_key: string | null;
|
|
deleted_at: Date | string | null;
|
|
}>(
|
|
`SELECT id, source_variant_key, deleted_at
|
|
FROM atlas.product_variants
|
|
WHERE product_id = $1
|
|
AND (source_variant_key = ANY($2::text[]) OR source_variant_key IS NULL)
|
|
ORDER BY sort_order, id`,
|
|
[productId, sourceKeys],
|
|
);
|
|
const existingBySourceKey = new Map(
|
|
existing.rows
|
|
.filter((variant) => variant.source_variant_key)
|
|
.map((variant) => [variant.source_variant_key!, variant]),
|
|
);
|
|
// A capture without a stable source key has exactly one synthetic
|
|
// default SKU. Reuse an existing unkeyed row when possible; this keeps
|
|
// its platform mapping rather than attempting a duplicate insert.
|
|
const existingDefault = existing.rows.find(
|
|
(variant) => variant.source_variant_key === null,
|
|
);
|
|
const variants = new Map<string, string>();
|
|
const retainedVariantIds: string[] = [];
|
|
for (let index = 0; index < variantRows.length; index += 1) {
|
|
const variant = variantRows[index];
|
|
const matched = variant.sourceVariantKey
|
|
? existingBySourceKey.get(variant.sourceVariantKey)
|
|
: existingDefault;
|
|
let variantId: string;
|
|
if (matched) {
|
|
variantId = matched.id;
|
|
// A user may have deliberately removed an Atlas SKU after
|
|
// editing began. Keep that decision; its source evidence is
|
|
// still present in normalized_payload, but it is not silently
|
|
// revived into the editable/publishable SKU set.
|
|
if (!preserveEditorDraft) {
|
|
await tx.query(
|
|
`UPDATE atlas.product_variants
|
|
SET source_sku = $1, display_name = $2, sort_order = $3,
|
|
deleted_at = NULL, updated_at = now()
|
|
WHERE id = $4`,
|
|
[
|
|
variant.sourceSku ?? null,
|
|
variant.displayName,
|
|
index,
|
|
variantId,
|
|
],
|
|
);
|
|
}
|
|
} else {
|
|
const inserted = await tx.query<{ id: string }>(
|
|
`INSERT INTO atlas.product_variants
|
|
(workspace_id, product_id, source_variant_key, source_sku, display_name, sort_order)
|
|
VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
|
|
[
|
|
workspaceId,
|
|
productId,
|
|
variant.sourceVariantKey,
|
|
variant.sourceSku ?? null,
|
|
variant.displayName,
|
|
index,
|
|
],
|
|
);
|
|
variantId = inserted.rows[0].id;
|
|
await this.ensureProfileListingsForVariant(
|
|
tx,
|
|
productId,
|
|
variantId,
|
|
);
|
|
}
|
|
retainedVariantIds.push(variantId);
|
|
if (variant.sourceVariantKey)
|
|
variants.set(variant.sourceVariantKey, variantId);
|
|
}
|
|
// Before an editor profile exists, variants are still just the latest
|
|
// source projection, so a missing source SKU can be safely hidden. Do
|
|
// not physically delete it: the partial source-key unique index does
|
|
// not exclude soft-deleted rows, and a later refresh may legitimately
|
|
// rediscover the same stable key. After editing begins, omission from a
|
|
// source refresh must never act as a user-requested SKU deletion.
|
|
if (!preserveEditorDraft) {
|
|
await tx.query(
|
|
`UPDATE atlas.product_variants
|
|
SET deleted_at = now(), updated_at = now()
|
|
WHERE product_id = $1 AND deleted_at IS NULL
|
|
AND NOT (id = ANY($2::uuid[]))`,
|
|
[productId, retainedVariantIds],
|
|
);
|
|
}
|
|
for (const asset of payload.assets) {
|
|
await tx.query(
|
|
`INSERT INTO atlas.collection_asset_transfers (workspace_id, collection_item_id, product_id, source_variant_id, media_type, poster_url, role, source_url, status, error_payload, source_snapshot_revision)
|
|
VALUES ($1, $2, $3, $4, $5::atlas.collection_media_type, $6, $7::atlas.media_candidate_role, $8, 'pending_selection', $9, $10)`,
|
|
[
|
|
workspaceId,
|
|
collectionItemId,
|
|
productId,
|
|
asset.variantSourceKey
|
|
? (variants.get(asset.variantSourceKey) ?? null)
|
|
: null,
|
|
asset.mediaType,
|
|
asset.posterUrl ?? null,
|
|
asset.role,
|
|
asset.url,
|
|
JSON.stringify({
|
|
relation: asset.relation,
|
|
referer_origin: asset.refererOrigin ?? null,
|
|
evidence_ref: asset.evidenceRef ?? null,
|
|
sort_order: asset.sortOrder,
|
|
}),
|
|
snapshotRevision,
|
|
],
|
|
);
|
|
}
|
|
// Collecting records source-media candidates only. A later explicit
|
|
// selection queues the media and creates the materialisation event.
|
|
await tx.query(
|
|
`UPDATE atlas.collection_items SET status = CASE WHEN status = 'needs_review' THEN status ELSE 'ready_for_edit' END, updated_at = now() WHERE id = $1 AND source_snapshot_revision = $2`,
|
|
[collectionItemId, snapshotRevision],
|
|
);
|
|
}
|
|
|
|
private async ensureProfileListingsForVariant(
|
|
tx: { query: Database['query'] },
|
|
productId: string,
|
|
variantId: string,
|
|
) {
|
|
// A source refresh may discover a new real SKU after a user has
|
|
// already entered one or more platform editors. Every existing profile
|
|
// needs a fresh, explicitly not-listed mapping for that SKU; imported
|
|
// mappings for other SKU rows are untouched.
|
|
await tx.query(
|
|
`INSERT INTO atlas.product_variant_platform_listings
|
|
(workspace_id, product_id, profile_id, platform_id, product_variant_id, store_id, currency_code)
|
|
SELECT current_setting('app.workspace_id')::uuid, $1, p.id, p.platform_id,
|
|
$2, p.target_store_id, s.currency_code
|
|
FROM atlas.product_platform_profiles p
|
|
JOIN atlas.stores s ON s.id = p.target_store_id
|
|
WHERE p.product_id = $1
|
|
ON CONFLICT (workspace_id, profile_id, product_variant_id) DO NOTHING`,
|
|
[productId, variantId],
|
|
);
|
|
}
|
|
|
|
private async materialResponse(
|
|
tx: { query: Database['query'] },
|
|
collectionItemId: string,
|
|
productId: string,
|
|
reused: boolean,
|
|
refreshed: boolean,
|
|
) {
|
|
const count = await tx.query<{
|
|
total: string;
|
|
pending_selection: string;
|
|
queued: string;
|
|
succeeded: string;
|
|
failed: string;
|
|
}>(
|
|
`SELECT count(*)::text AS total, count(*) FILTER (WHERE t.status = 'pending_selection')::text AS pending_selection, count(*) FILTER (WHERE t.status = 'queued')::text AS queued, count(*) FILTER (WHERE t.status = 'succeeded')::text AS succeeded, count(*) FILTER (WHERE t.status = 'failed')::text AS failed FROM atlas.collection_asset_transfers t JOIN atlas.collection_items c ON c.id = t.collection_item_id WHERE t.collection_item_id = $1 AND t.source_snapshot_revision = c.source_snapshot_revision`,
|
|
[collectionItemId],
|
|
);
|
|
const status = await tx.query<{
|
|
status: string;
|
|
source_title: string | null;
|
|
}>(
|
|
'SELECT status, source_title FROM atlas.collection_items WHERE id = $1',
|
|
[collectionItemId],
|
|
);
|
|
return {
|
|
collection_item: {
|
|
id: collectionItemId,
|
|
status: status.rows[0]?.status ?? 'received',
|
|
source_title: status.rows[0]?.source_title ?? null,
|
|
variant_count: 0,
|
|
},
|
|
product_id: productId,
|
|
reused,
|
|
refreshed,
|
|
editor_entry_path: `/collection-test/${collectionItemId}`,
|
|
materialization: {
|
|
total: Number(count.rows[0]?.total ?? 0),
|
|
pending_selection: Number(
|
|
count.rows[0]?.pending_selection ?? 0,
|
|
),
|
|
queued: Number(count.rows[0]?.queued ?? 0),
|
|
succeeded: Number(count.rows[0]?.succeeded ?? 0),
|
|
failed: Number(count.rows[0]?.failed ?? 0),
|
|
},
|
|
};
|
|
}
|
|
}
|
|
|
|
function sourceState(payload: CollectedProductPayload) {
|
|
const verifiedVariants = payload.variants.filter(
|
|
(variant) => variant.sourceVariantKey.trim().length > 0,
|
|
);
|
|
return {
|
|
status:
|
|
verifiedVariants.length || !payload.variantAxes.length
|
|
? 'ready_for_edit'
|
|
: 'needs_review',
|
|
sourcePrice: payload.common.sourcePrice,
|
|
};
|
|
}
|