joey5628/atlas-seller-kit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Expand seller store functionality

Browse Source
This commit is contained in:
R524809
2026-09-07 17:51:11 +08:00
parent 4c0f3a4a79
commit ceb644a3f9
35 changed files with 32883 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/extension/extension.service.ts
+131 -108
View File
@@ -1,140 +1,163 @@
import { createHash, randomUUID } from 'node:crypto';
import { Inject, Injectable } from '@nestjs/common';
import type { CollectedProductPayload } from '@atlas/contracts';
import type { Database } from '@atlas/database';
import { createOpaqueToken, signExtensionAccessToken, tokenHash, verifyExtensionAccessToken } from '../auth/auth.crypto.js';
import { AuthService } from '../auth/auth.service.js';
import { ApiError } from '../common/api-error.js';
import { IdempotencyService } from '../common/idempotency.service.js';
import { APP_CONFIG, type AppConfig } from '../config/app-config.module.js';
import { DATABASE } from '../database/database.module.js';
type AuthorizeInput = { client_id: string; redirect_uri: string; state: string; code_challenge: string; code_challenge_method: 'S256' };
type ExchangeInput = { client_id: string; code: string; redirect_uri: string; code_verifier: string };
type ExtensionSession = { id: string; workspace_id: string; user_id: string; client_id: string; expires_at: Date | string; revoked_at: Date | string | null };
@Injectable()
export class ExtensionService {
public constructor(
@Inject(DATABASE) private readonly database: Database,
@Inject(AuthService) private readonly auth: AuthService,
@Inject(IdempotencyService) private readonly idempotency: IdempotencyService,
@Inject(APP_CONFIG) private readonly config: AppConfig,
) {}
public async authorize(input: AuthorizeInput, refreshToken: string | undefined) {
this.assertClient(input.client_id, input.redirect_uri);
const session = await this.auth.resolveWebRefreshSession(refreshToken);
if (!session) return { login_url: `${this.config.WEB_ORIGIN}/login?continue=${encodeURIComponent(this.authorizationUrl(input))}` };
// Prefixing a high-entropy secret with the workspace UUID lets the server
// establish RLS before looking up the code; it is not an authorization secret.
const rawCode = `${session.workspace.id}.${createOpaqueToken()}`;
await this.database.workspaceTransaction(session.workspace.id, async (tx) => {
await tx.query(`INSERT INTO atlas.extension_authorization_codes
(workspace_id, user_id, client_id, redirect_uri, code_hash, code_challenge, expires_at)
VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4, $5, now() + interval '60 seconds')`,
[session.userId, input.client_id, input.redirect_uri, tokenHash(rawCode), input.code_challenge]);
});
const callback = new URL(input.redirect_uri);
callback.searchParams.set('code', rawCode);
callback.searchParams.set('state', input.state);
return { redirect_url: callback.toString() };
}
public async exchange(input: ExchangeInput) {
this.assertClient(input.client_id, input.redirect_uri);
const hash = tokenHash(input.code); const workspaceId = workspacePrefix(input.code);
if (!workspaceId) throw new ApiError(401, 'EXTENSION_AUTHORIZATION_INVALID', '扩展授权码无效或已过期');
const code = await this.database.workspaceTransaction(workspaceId, async (tx) => (await tx.query<{ workspace_id: string; user_id: string; client_id: string; redirect_uri: string; code_challenge: string }>(
`SELECT workspace_id, user_id, client_id, redirect_uri, code_challenge FROM atlas.extension_authorization_codes WHERE code_hash = $1 AND consumed_at IS NULL AND expires_at > now()`, [hash])).rows[0]);
if (!code || code.client_id !== input.client_id || code.redirect_uri !== input.redirect_uri || !constantEquals(code.code_challenge, pkceChallenge(input.code_verifier))) {
throw new ApiError(401, 'EXTENSION_AUTHORIZATION_INVALID', '扩展授权码无效或已过期');
}
return this.database.workspaceTransaction(code.workspace_id, async (tx) => {
const consumed = await tx.query(`UPDATE atlas.extension_authorization_codes SET consumed_at = now()
WHERE code_hash = $1 AND consumed_at IS NULL AND expires_at > now()`, [hash]);
if (!consumed.rowCount) throw new ApiError(401, 'EXTENSION_AUTHORIZATION_INVALID', '扩展授权码无效或已过期');
return this.createSession(tx, code.workspace_id, code.user_id, code.client_id);
});
}
public async refresh(refreshToken: string) {
const workspaceId = workspacePrefix(refreshToken); if (!workspaceId) throw new ApiError(401, 'EXTENSION_SESSION_EXPIRED', '扩展登录已过期');
return this.database.workspaceTransaction(workspaceId, async (tx) => {
const result = await tx.query<ExtensionSession>(`SELECT id, workspace_id, user_id, client_id, expires_at, revoked_at FROM atlas.extension_sessions WHERE refresh_token_hash = $1 AND revoked_at IS NULL AND expires_at > now() FOR UPDATE`, [tokenHash(refreshToken)]);
const old = result.rows[0];
if (!old) throw new ApiError(401, 'EXTENSION_SESSION_EXPIRED', '扩展登录已过期');
await tx.query('UPDATE atlas.extension_sessions SET revoked_at = now(), last_used_at = now() WHERE id = $1', [old.id]);
return this.createSession(tx, old.workspace_id, old.user_id, old.client_id);
});
}
public async revoke(workspaceId: string, sessionId: string) {
await this.database.workspaceTransaction(workspaceId, (tx) => tx.query('UPDATE atlas.extension_sessions SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL', [sessionId]));
}
public async authenticate(header: string | undefined) {
const token = header?.replace(/^Bearer\s+/i, '');
const claims = token ? verifyExtensionAccessToken(token, this.config.AUTH_JWT_SECRET) : null;
if (!claims) throw new ApiError(401, 'EXTENSION_SESSION_EXPIRED', '扩展登录已过期');
const session = await this.database.workspaceTransaction(claims.wid, async (tx) => (await tx.query<ExtensionSession>(`SELECT id, workspace_id, user_id, client_id, expires_at, revoked_at FROM atlas.extension_sessions WHERE id = $1 AND user_id = $2 AND revoked_at IS NULL AND expires_at > now()`, [claims.sid, claims.sub])).rows[0]);
if (!session) throw new ApiError(401, 'EXTENSION_SESSION_EXPIRED', '扩展登录已过期');
return { userId: session.user_id, workspaceId: session.workspace_id, sessionId: session.id, clientId: session.client_id };
}
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 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`, [platform.rows[0].id, payload.source.itemKey]);
if (existing.rows[0]) return { status: 200, body: { data: await this.materialResponse(tx, existing.rows[0].id, existing.rows[0].product_id, true) } };
const verifiedVariants = payload.variants.filter((variant) => variant.sourceVariantKey.trim().length > 0);
const status = verifiedVariants.length || !payload.variantAxes.length ? 'materializing' : 'needs_review';
const sourcePrice = payload.common.sourcePrice;
const existing = await this.findExistingSourceItem(
tx,
workspaceId,
platform.rows[0].id,
payload,
);
if (existing) {
await this.replaceSourceSnapshot(
tx,
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]);
const variantRows = verifiedVariants.length ? verifiedVariants : [{ sourceVariantKey: null, sourceSku: undefined, displayName: '默认 SKU' }];
const variants = new Map<string, string>();
for (let index = 0; index < variantRows.length; index += 1) {
const variant = variantRows[index];
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 (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4, $5) RETURNING id`, [product.rows[0].id, variant.sourceVariantKey, variant.sourceSku ?? null, variant.displayName, index]);
if (variant.sourceVariantKey) variants.set(variant.sourceVariantKey, inserted.rows[0].id);
}
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, role, source_url, status, error_payload)
VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4::atlas.media_candidate_role, $5, 'queued', $6)`,
[item.rows[0].id, product.rows[0].id, asset.variantSourceKey ? variants.get(asset.variantSourceKey) ?? null : null, asset.role, asset.url, JSON.stringify({ relation: asset.relation, referer_origin: asset.refererOrigin ?? null, evidence_ref: asset.evidenceRef ?? null, sort_order: asset.sortOrder })]);
}
if (payload.assets.length) await tx.query(`INSERT INTO atlas.outbox_events (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key)
VALUES (current_setting('app.workspace_id')::uuid, 'collection_item', $1, 'collection.assets.materialize.requested', jsonb_build_object('collection_item_id', $1), $2)`, [item.rows[0].id, `collection.assets.materialize:${item.rows[0].id}`]);
else 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`, [item.rows[0].id]);
return { status: 201, body: { data: await this.materialResponse(tx, item.rows[0].id, product.rows[0].id, false) } };
await this.writeSourceSnapshotRows(tx, item.rows[0].id, product.rows[0].id, 1, payload);
return { status: 201, body: { data: await this.materialResponse(tx, item.rows[0].id, product.rows[0].id, false, false) } };
});
}
private async materialResponse(tx: { query: Database['query'] }, collectionItemId: string, productId: string, reused: boolean) {
const count = await tx.query<{ total: string; queued: string; succeeded: string; failed: string }>(`SELECT count(*)::text AS total, count(*) FILTER (WHERE status = 'queued')::text AS queued, count(*) FILTER (WHERE status = 'succeeded')::text AS succeeded, count(*) FILTER (WHERE status = 'failed')::text AS failed FROM atlas.collection_asset_transfers WHERE collection_item_id = $1`, [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, editor_entry_path: `/collection-test/${collectionItemId}`, materialization: { total: Number(count.rows[0]?.total ?? 0), queued: Number(count.rows[0]?.queued ?? 0), succeeded: Number(count.rows[0]?.succeeded ?? 0), failed: Number(count.rows[0]?.failed ?? 0) } };
/**
* 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 createSession(tx: { query: Database['query'] }, workspaceId: string, userId: string, clientId: string) {
const refreshToken = `${workspaceId}.${createOpaqueToken()}`; const id = randomUUID();
await tx.query(`INSERT INTO atlas.extension_sessions (id, workspace_id, user_id, client_id, refresh_token_hash, expires_at)
VALUES ($1, current_setting('app.workspace_id')::uuid, $2, $3, $4, now() + interval '30 days')`, [id, userId, clientId, tokenHash(refreshToken)]);
const accessToken = signExtensionAccessToken({ sub: userId, sid: id, wid: workspaceId, typ: 'extension', iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 15 * 60 }, this.config.AUTH_JWT_SECRET);
return { status: 200, body: { data: { access_token: accessToken, expires_in: 900, refresh_token: refreshToken, refresh_token_expires_in: 2_592_000, token_type: 'Bearer' } } };
private async replaceSourceSnapshot(tx: { query: Database['query'] }, collectionItemId: string, productId: string, payload: CollectedProductPayload) {
const { status, sourcePrice } = sourceState(payload);
// 不删除旧任务记录：运行中的 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 ('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('采集来源快照版本更新失败');
await tx.query(`UPDATE atlas.products SET source_title = $1, updated_at = now(), version = version + 1 WHERE id = $2`, [payload.common.title?.value ?? null, productId]);
await tx.query('DELETE FROM atlas.product_variants WHERE product_id = $1', [productId]);
await this.writeSourceSnapshotRows(tx, collectionItemId, productId, snapshotRevision, payload);
return snapshotRevision;
}
private assertClient(clientId: string, redirectUri: string) { if (clientId !== this.config.EXTENSION_COLLECTOR_CLIENT_ID || redirectUri !== this.config.EXTENSION_COLLECTOR_REDIRECT_URI) throw new ApiError(422, 'EXTENSION_CLIENT_INVALID', '扩展客户端或回调地址未登记'); }
private authorizationUrl(input: AuthorizeInput) { const url = new URL(`${this.config.API_PUBLIC_ORIGIN}/api/extension/authorize`); for (const [key, value] of Object.entries(input)) url.searchParams.set(key, value); return url.toString(); }
private async writeSourceSnapshotRows(tx: { query: Database['query'] }, collectionItemId: string, productId: string, snapshotRevision: number, payload: CollectedProductPayload) {
const verifiedVariants = payload.variants.filter((variant) => variant.sourceVariantKey.trim().length > 0);
const variantRows = verifiedVariants.length ? verifiedVariants : [{ sourceVariantKey: null, sourceSku: undefined, displayName: '默认 SKU' }];
const variants = new Map<string, string>();
for (let index = 0; index < variantRows.length; index += 1) {
const variant = variantRows[index];
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 (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4, $5) RETURNING id`, [productId, variant.sourceVariantKey, variant.sourceSku ?? null, variant.displayName, index]);
if (variant.sourceVariantKey) variants.set(variant.sourceVariantKey, inserted.rows[0].id);
}
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, role, source_url, status, error_payload, source_snapshot_revision)
VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4::atlas.media_candidate_role, $5, 'queued', $6, $7)`,
[collectionItemId, productId, asset.variantSourceKey ? variants.get(asset.variantSourceKey) ?? null : 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]);
}
if (payload.assets.length) await tx.query(`INSERT INTO atlas.outbox_events (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key)
VALUES (current_setting('app.workspace_id')::uuid, 'collection_item', $1, 'collection.assets.materialize.requested', jsonb_build_object('collection_item_id', $1::uuid, 'source_snapshot_revision', $2::bigint), $3)`, [collectionItemId, snapshotRevision, `collection.assets.materialize:${collectionItemId}:${snapshotRevision}`]);
else 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 materialResponse(tx: { query: Database['query'] }, collectionItemId: string, productId: string, reused: boolean, refreshed: boolean) {
const count = await tx.query<{ total: string; queued: string; succeeded: string; failed: string }>(`SELECT count(*)::text AS total, 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), 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 ? 'materializing' : 'needs_review', sourcePrice: payload.common.sourcePrice };
}
function pkceChallenge(verifier: string) { return createHash('sha256').update(verifier).digest('base64url'); }
function constantEquals(left: string, right: string) { return left.length === right.length && left.split('').every((value, index) => value === right[index]); }
function workspacePrefix(value: string) { const prefix = value.split('.', 1)[0]; return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(prefix) ? prefix : null; }