From 70d9d5b92a7d2bd4d47eda91fe1833465ffbc83b Mon Sep 17 00:00:00 2001 From: R524809 Date: Fri, 4 Sep 2026 10:27:29 +0800 Subject: [PATCH] =?UTF-8?q?feat(store):=20=E6=B7=BB=E5=8A=A0=E9=97=A8?= =?UTF-8?q?=E5=BA=97=E3=80=81=E9=97=A8=E5=BA=97=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Atlas Local - Migrator.session.sql | 0 README.md | 4 +- apps/api/src/stores/platforms.service.ts | 4 + apps/api/src/stores/store-secret.crypto.ts | 21 ++ apps/api/src/stores/stores.service.ts | 29 ++- apps/web/src/pages/shops/ShopsPage.tsx | 10 +- apps/worker/package.json | 3 + apps/worker/src/check-ozon-store.ts | 70 ++++++ apps/worker/src/config.ts | 23 ++ apps/worker/src/main.ts | 210 ++++++++++++++++-- apps/worker/src/outbox-dispatcher.ts | 94 ++++++++ apps/worker/src/ozon-client.ts | 46 ++++ apps/worker/src/store-secret.crypto.ts | 20 ++ docs/api/stores-and-platforms.md | 29 ++- docs/guides/连接数据库.md | 26 +++ docs/project/HANDOFF.md | 19 +- packages/contracts/src/stores.ts | 2 + .../005_ozon_cny_default_currency.sql | 15 ++ pnpm-lock.yaml | 3 + 19 files changed, 602 insertions(+), 26 deletions(-) create mode 100644 Atlas Local - Migrator.session.sql create mode 100644 apps/worker/src/check-ozon-store.ts create mode 100644 apps/worker/src/config.ts create mode 100644 apps/worker/src/outbox-dispatcher.ts create mode 100644 apps/worker/src/ozon-client.ts create mode 100644 apps/worker/src/store-secret.crypto.ts create mode 100644 docs/guides/连接数据库.md create mode 100644 packages/database/migrations/005_ozon_cny_default_currency.sql diff --git a/Atlas Local - Migrator.session.sql b/Atlas Local - Migrator.session.sql new file mode 100644 index 0000000..e69de29 diff --git a/README.md b/README.md index 6776a79..6804ac2 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,9 @@ Atlas Seller Kit 是面向多用户、多门店、多平台与多国家市场的 ## 当前状态 -仅完成产品与架构设计，尚未开始应用代码开发。开始新 AI 对话或开发前，请先阅读 [项目交接](./docs/project/HANDOFF.md)，再阅读 [文档导航](./docs/README.md)。 +项目已完成 monorepo 与本地基础设施初始化，并实现认证、个人工作空间、平台目录/上下文和门店管理 HTTP API。Web 已提供登录/注册、主布局、页头平台选择器和接入真实 API 的店铺管理页；其余业务页面仍为占位。门店凭据验证已接入 Outbox Dispatcher 与 Worker；仓库同步仍待实现。 + +开始新 AI 对话或开发前，请先阅读 [项目交接](./docs/project/HANDOFF.md)，再阅读 [文档导航](./docs/README.md)。 ## 设计目标 diff --git a/apps/api/src/stores/platforms.service.ts b/apps/api/src/stores/platforms.service.ts index 6fde9df..3368148 100644 --- a/apps/api/src/stores/platforms.service.ts +++ b/apps/api/src/stores/platforms.service.ts @@ -41,6 +41,10 @@ export class PlatformsService { platform.capabilities, 'default_locale', ), + default_currency_code: stringCapability( + platform.capabilities, + 'default_currency_code', + ), })); } diff --git a/apps/api/src/stores/store-secret.crypto.ts b/apps/api/src/stores/store-secret.crypto.ts index f442a51..25f2af8 100644 --- a/apps/api/src/stores/store-secret.crypto.ts +++ b/apps/api/src/stores/store-secret.crypto.ts @@ -1,4 +1,5 @@ import { + createDecipheriv, createCipheriv, createHash, createHmac, @@ -35,3 +36,23 @@ export function encryptStoreSecret( .digest('base64url'), }; } + +/** 仅供受信任的后台 Worker 使用；调用方不得记录返回值。 */ +export function decryptStoreSecret( + ciphertextWithTag: Buffer, + nonce: Buffer, + keyMaterial: string, +): string { + const authenticationTagLength = 16; + if (ciphertextWithTag.length <= authenticationTagLength) { + throw new Error('Invalid encrypted store secret'); + } + const key = createHash('sha256').update(keyMaterial).digest(); + const ciphertext = ciphertextWithTag.subarray(0, -authenticationTagLength); + const tag = ciphertextWithTag.subarray(-authenticationTagLength); + const decipher = createDecipheriv('aes-256-gcm', key, nonce); + decipher.setAuthTag(tag); + return Buffer.concat([decipher.update(ciphertext), decipher.final()]).toString( + 'utf8', + ); +} diff --git a/apps/api/src/stores/stores.service.ts b/apps/api/src/stores/stores.service.ts index a30e3c1..ceb7f73 100644 --- a/apps/api/src/stores/stores.service.ts +++ b/apps/api/src/stores/stores.service.ts @@ -470,6 +470,19 @@ export class StoresService { '门店凭据尚未验证通过，无法同步仓库', ); } + if (kind === 'store_credential_verification') { + await transaction.query( + `UPDATE atlas.stores SET status = 'verifying', credential_status = 'unknown', + last_error_code = NULL, last_error_message = NULL, last_error_at = NULL, + revision = revision + 1, updated_at = now() WHERE id = $1`, + [storeId], + ); + await transaction.query( + `UPDATE atlas.store_secrets SET status = 'unknown', updated_at = now() + WHERE store_id = $1 AND kind = 'seller_api_key' AND revoked_at IS NULL`, + [storeId], + ); + } const operation = await this.createOperation( transaction, userId, @@ -605,6 +618,20 @@ export class StoresService { [kind, storeId, userId], ); const operation = toOperation(created.rows[0]); + const payload: { store_id: string; operation_id: string; secret_id?: string } = { + store_id: storeId, + operation_id: operation.id, + }; + // 验证任务绑定创建当刻的密钥行；密钥后来被替换时，旧任务不得验证或覆盖新状态。 + if (kind === 'store_credential_verification') { + const secret = await transaction.query<{ id: string }>( + `SELECT id FROM atlas.store_secrets + WHERE store_id = $1 AND kind = 'seller_api_key' AND revoked_at IS NULL`, + [storeId], + ); + if (!secret.rows[0]) throw new Error('Current store secret is missing'); + payload.secret_id = secret.rows[0].id; + } await transaction.query( `INSERT INTO atlas.outbox_events (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key) @@ -612,7 +639,7 @@ export class StoresService { [ storeId, `${kind}.requested`, - { store_id: storeId, operation_id: operation.id }, + payload, operation.id, ], ); diff --git a/apps/web/src/pages/shops/ShopsPage.tsx b/apps/web/src/pages/shops/ShopsPage.tsx index 4639d8a..3572929 100644 --- a/apps/web/src/pages/shops/ShopsPage.tsx +++ b/apps/web/src/pages/shops/ShopsPage.tsx @@ -157,7 +157,11 @@ const ShopsPage = () => { const openCreate = () => { createForm.resetFields(); - createForm.setFieldsValue({ is_default: stores.length === 0 }); + createForm.setFieldsValue({ + is_default: stores.length === 0, + currency_code: + currentPlatform?.default_currency_code ?? currencyOptions[0]?.value, + }); setCreateOpen(true); }; @@ -347,7 +351,9 @@ const ShopsPage = () => { return ( {status.text} - {credential.text} + {store.status !== 'verifying' && ( + {credential.text} + )} {store.last_error && ( '); +} + +const database = createApplicationDatabase(workerConfig.databaseUrl); + +async function main() { + const workspaces = await database.query<{ id: string }>( + 'SELECT id FROM atlas.workspaces', + ); + const matches: StoreCredential[] = []; + for (const workspace of workspaces.rows) { + const found = await database.workspaceTransaction(workspace.id, (tx) => + tx.query( + `SELECT s.workspace_id, s.id AS store_id, s.platform_client_id AS client_id, + secret.secret_ciphertext AS ciphertext, secret.secret_nonce AS nonce + FROM atlas.stores s + JOIN atlas.platforms p ON p.id = s.platform_id AND p.code = 'OZON' + JOIN atlas.store_secrets secret ON secret.store_id = s.id + AND secret.kind = 'seller_api_key' AND secret.revoked_at IS NULL + WHERE s.name = $1 AND s.deleted_at IS NULL`, + [storeName], + ), + ); + matches.push(...found.rows); + } + if (matches.length === 0) throw new Error(`No active Ozon store named “${storeName}”`); + if (matches.length > 1) throw new Error(`Multiple active Ozon stores named “${storeName}”; use a unique name`); + + const store = matches[0]; + const apiKey = decryptStoreSecret(store.ciphertext, store.nonce, workerConfig.storeSecretEncryptionKey); + const checks = [ + { name: 'warehouse-list', path: '/v1/warehouse/list', body: {} }, + { name: 'category-tree', path: '/v1/description-category/tree', body: { language: 'ZH_HANS' } }, + ]; + for (const check of checks) { + let response: Response; + try { + response = await fetch(`https://api-seller.ozon.ru${check.path}`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Client-Id': store.client_id, + 'Api-Key': apiKey, + }, + body: JSON.stringify(check.body), + signal: AbortSignal.timeout(30_000), + }); + } catch { + console.info(`${check.name}: network error or timeout`); + continue; + } + console.info(`${check.name}: HTTP ${response.status} ${response.ok ? 'OK' : 'FAILED'}`); + } +} + +void main().finally(() => database.close()); diff --git a/apps/worker/src/config.ts b/apps/worker/src/config.ts new file mode 100644 index 0000000..04a5c97 --- /dev/null +++ b/apps/worker/src/config.ts @@ -0,0 +1,23 @@ +import { existsSync } from 'node:fs'; +import { resolve } from 'node:path'; +import { loadEnvFile } from 'node:process'; + +for (const directory of [process.cwd(), resolve(process.cwd(), '..'), resolve(process.cwd(), '../..')]) { + const path = resolve(directory, '.env'); + if (existsSync(path)) { + loadEnvFile(path); + break; + } +} + +function required(name: string): string { + const value = process.env[name]; + if (!value) throw new Error(`${name} is required`); + return value; +} + +export const workerConfig = { + databaseUrl: required('DATABASE_URL'), + redisUrl: required('REDIS_URL'), + storeSecretEncryptionKey: required('STORE_SECRET_ENCRYPTION_KEY'), +}; diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index bf938d1..4fd5953 100644 --- a/apps/worker/src/main.ts +++ b/apps/worker/src/main.ts @@ -1,28 +1,202 @@ -import { Queue } from 'bullmq'; -import IORedis from 'ioredis'; +import { randomUUID } from 'node:crypto'; +import { Queue, Worker } from 'bullmq'; +import { Redis } from 'ioredis'; +import { createApplicationDatabase, type Database } from '@atlas/database'; +import { workerConfig } from './config.js'; +import { OutboxDispatcher, type OutboxJob } from './outbox-dispatcher.js'; +import { OzonCredentialError, OzonRequestError, verifyOzonCredentials } from './ozon-client.js'; +import { decryptStoreSecret } from './store-secret.crypto.js'; -// 显式构造客户端实例传给 BullMQ：原生 ESM（pnpm start）下 BullMQ 无法按 connection -// 选项自动加载 ioredis；maxRetriesPerRequest: null 是 BullMQ Worker 的要求 -const connection = new IORedis( - process.env.REDIS_URL ?? 'redis://localhost:6379', - { - maxRetriesPerRequest: null, +const connection = new Redis(workerConfig.redisUrl, { maxRetriesPerRequest: null }); +const database = createApplicationDatabase(workerConfig.databaseUrl); +const queue = new Queue('atlas-dispatch', { connection }); +const dispatcher = new OutboxDispatcher(database, queue, `worker_${randomUUID()}`); + +const worker = new Worker( + 'atlas-dispatch', + async (job) => { + if (job.name === 'store_credential_verification.requested') { + await verifyStoreCredentials(database, job.data); + } }, + { connection, concurrency: 5 }, ); -const dispatchQueue = new Queue('atlas-dispatch', { connection }); +worker.on('failed', (job, error) => { + if (!job || job.attemptsMade < (job.opts.attempts ?? 1)) return; + void markUnavailable(database, job.data, error); +}); + +async function verifyStoreCredentials(database: Database, job: OutboxJob) { + const source = await 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 result = await tx.query<{ + platform_code: string; platform_client_id: string; secret_ciphertext: Buffer; secret_nonce: Buffer; + }>( + `SELECT p.code AS platform_code, s.platform_client_id, secret.secret_ciphertext, secret.secret_nonce + FROM atlas.operations operation JOIN atlas.stores s ON s.id = operation.resource_id AND s.deleted_at IS NULL + JOIN atlas.platforms p ON p.id = s.platform_id JOIN atlas.store_secrets secret ON secret.id = $2 + WHERE operation.id = $1 AND operation.kind = 'store_credential_verification' + AND secret.store_id = s.id AND secret.kind = 'seller_api_key' AND secret.revoked_at IS NULL`, + [job.payload.operation_id, job.payload.secret_id ?? ''], + ); + return result.rows[0] ?? null; + }); + if (!source) return cancelStaleOperation(database, job); + const apiKey = decryptStoreSecret(source.secret_ciphertext, source.secret_nonce, workerConfig.storeSecretEncryptionKey); + try { + if (source.platform_code.toUpperCase() !== 'OZON') throw new OzonRequestError(422); + await verifyOzonCredentials(source.platform_client_id, apiKey); + await completeVerification(database, job, true); + } catch (error) { + if (error instanceof OzonCredentialError) { + await completeVerification(database, job, false, error.status); + return; + } + if (error instanceof OzonRequestError) { + await markRequestRejected(database, job, error.status); + return; + } + throw error; + } +} + +async function cancelStaleOperation(database: Database, job: OutboxJob) { + await database.workspaceTransaction(job.workspaceId, (tx) => tx.query( + `UPDATE atlas.operations SET status = 'cancelled', completed_at = now(), + error_code = 'CREDENTIAL_REPLACED', error_message = '凭据已更新，旧验证任务已取消' + WHERE id = $1 AND status IN ('queued', 'running')`, + [job.payload.operation_id], + )); +} + +async function completeVerification( + database: Database, + job: OutboxJob, + valid: boolean, + rejectedStatus?: 401 | 403, +) { + await database.workspaceTransaction(job.workspaceId, async (tx) => { + const current = await tx.query<{ id: string }>( + `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2 + WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id + AND secret.revoked_at IS NULL FOR UPDATE`, + [job.payload.store_id, job.payload.secret_id ?? ''], + ); + if (!current.rows[0]) return; + const state = valid ? 'valid' : 'invalid'; + await tx.query( + `UPDATE atlas.store_secrets SET status = $1::atlas.credential_status, + last_verified_at = now(), updated_at = now() WHERE id = $2`, + [state, job.payload.secret_id], + ); + await tx.query( + valid + ? `UPDATE atlas.stores SET status = 'active', credential_status = 'valid', + credential_checked_at = now(), last_error_code = NULL, last_error_message = NULL, + last_error_at = NULL, revision = revision + 1, updated_at = now() WHERE id = $1` + : `UPDATE atlas.stores SET status = 'credential_invalid', credential_status = 'invalid', + credential_checked_at = now(), last_error_code = $2, + last_error_message = $3, last_error_at = now(), + revision = revision + 1, updated_at = now() WHERE id = $1`, + valid ? [job.payload.store_id] : [ + job.payload.store_id, + rejectedStatus === 403 ? 'OZON_CREDENTIAL_FORBIDDEN' : 'OZON_CREDENTIAL_UNAUTHORIZED', + rejectedStatus === 403 + ? 'Ozon 拒绝访问：请检查 Client ID 与 API Key 是否属于同一账户，及密钥是否具备读取权限' + : 'Ozon 未接受该 Client ID 或 API Key', + ], + ); + await tx.query( + valid + ? `UPDATE atlas.operations SET status = 'succeeded', completed_at = now(), + result = jsonb_build_object('credential_status', 'valid'), error_code = NULL, error_message = NULL + WHERE id = $1 AND status IN ('queued', 'running')` + : `UPDATE atlas.operations SET status = 'failed', completed_at = now(), + error_code = $2, error_message = $3 + WHERE id = $1 AND status IN ('queued', 'running')`, + valid ? [job.payload.operation_id] : [ + job.payload.operation_id, + rejectedStatus === 403 ? 'OZON_CREDENTIAL_FORBIDDEN' : 'OZON_CREDENTIAL_UNAUTHORIZED', + rejectedStatus === 403 + ? 'Ozon 拒绝访问：请检查 Client ID 与 API Key 是否属于同一账户，及密钥是否具备读取权限' + : 'Ozon 未接受该 Client ID 或 API Key', + ], + ); + }); +} + +async function markRequestRejected(database: Database, job: OutboxJob, status: number) { + await database.workspaceTransaction(job.workspaceId, async (tx) => { + const current = await tx.query<{ id: string }>( + `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2 + WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id + AND secret.revoked_at IS NULL FOR UPDATE`, + [job.payload.store_id, job.payload.secret_id ?? ''], + ); + if (!current.rows[0]) return; + const message = `Ozon 拒绝了验证请求（HTTP ${status}）；这不代表 API Key 无效`; + await tx.query( + `UPDATE atlas.operations SET status = 'failed', completed_at = now(), + error_code = 'OZON_VERIFICATION_REQUEST_REJECTED', error_message = $2, + result = jsonb_build_object('ozon_http_status', $3) + WHERE id = $1 AND status IN ('queued', 'running')`, + [job.payload.operation_id, message, status], + ); + await tx.query( + `UPDATE atlas.stores SET status = 'sync_error', last_error_code = 'OZON_VERIFICATION_REQUEST_REJECTED', + last_error_message = $2, last_error_at = now(), revision = revision + 1, updated_at = now() + WHERE id = $1 AND status = 'verifying'`, + [job.payload.store_id, message], + ); + }); +} + +async function markUnavailable(database: Database, job: OutboxJob, _error: Error) { + await database.workspaceTransaction(job.workspaceId, async (tx) => { + const current = await tx.query<{ id: string }>( + `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2 + WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id + AND secret.revoked_at IS NULL FOR UPDATE`, + [job.payload.store_id, job.payload.secret_id ?? ''], + ); + if (!current.rows[0]) return; + await tx.query( + `UPDATE atlas.operations SET status = 'failed', completed_at = now(), + error_code = 'OZON_VERIFICATION_UNAVAILABLE', error_message = '暂时无法验证 Ozon 凭据，请稍后重试' + WHERE id = $1 AND status IN ('queued', 'running')`, + [job.payload.operation_id], + ); + await tx.query( + `UPDATE atlas.stores SET status = 'sync_error', last_error_code = 'OZON_VERIFICATION_UNAVAILABLE', + last_error_message = '暂时无法验证 Ozon 凭据，请稍后重试', last_error_at = now(), + revision = revision + 1, updated_at = now() WHERE id = $1 AND status = 'verifying'`, + [job.payload.store_id], + ); + }); +} + +let timer: ReturnType | undefined; async function bootstrap() { - await dispatchQueue.waitUntilReady(); - console.info( - 'Atlas Worker is ready; it will consume transactional-outbox events after that module exists.', - ); + await queue.waitUntilReady(); + await worker.waitUntilReady(); + await dispatcher.dispatchBatch(); + timer = setInterval(() => void dispatcher.dispatchBatch().catch((error: unknown) => { + console.error('outbox_dispatch_failed', error instanceof Error ? error.message : error); + }), 2_000); + console.info('Atlas Worker is ready; consuming transactional-outbox events.'); } - -void bootstrap(); - +void bootstrap().catch((error: unknown) => { + console.error('worker_startup_failed', error instanceof Error ? error.message : error); + process.exitCode = 1; +}); async function shutdown() { - await dispatchQueue.close(); + if (timer) clearInterval(timer); + await worker.close(); await queue.close(); await database.close(); await connection.quit(); } - process.once('SIGINT', () => void shutdown()); process.once('SIGTERM', () => void shutdown()); diff --git a/apps/worker/src/outbox-dispatcher.ts b/apps/worker/src/outbox-dispatcher.ts new file mode 100644 index 0000000..ba39ff6 --- /dev/null +++ b/apps/worker/src/outbox-dispatcher.ts @@ -0,0 +1,94 @@ +import type { Queue } from 'bullmq'; +import type { Database } from '@atlas/database'; + +export type OutboxJob = { + eventId: string; + workspaceId: string; + eventType: string; + payload: { store_id: string; operation_id: string; secret_id?: string }; +}; + +type EventRow = Omit & { + id: string; + workspace_id: string; + event_type: string; + event_payload: OutboxJob['payload']; +}; + +export class OutboxDispatcher { + public constructor( + private readonly database: Database, + private readonly queue: Queue, + private readonly workerId: string, + ) {} + + public async dispatchBatch() { + // 业务表受 RLS 保护；先枚举工作空间，再在各自租户事务中认领事件。 + const workspaces = await this.database.query<{ id: string }>( + 'SELECT id FROM atlas.workspaces', + ); + for (const workspace of workspaces.rows) { + const events = await this.claim(workspace.id); + for (const event of events) await this.dispatch(event); + } + } + + private async claim(workspaceId: string): Promise { + return this.database.workspaceTransaction(workspaceId, async (tx) => { + const events = await tx.query( + `WITH candidates AS ( + SELECT id FROM atlas.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.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.workspace_id, event.event_type, event.event_payload`, + [this.workerId], + ); + return events.rows.map((event) => ({ + eventId: event.id, + workspaceId: event.workspace_id, + eventType: event.event_type, + payload: event.event_payload, + })); + }); + } + + private async dispatch(event: OutboxJob) { + try { + await this.queue.add(event.eventType, event, { + jobId: event.eventId, + attempts: 5, + backoff: { type: 'exponential', delay: 5_000 }, + removeOnComplete: 1_000, + removeOnFail: 1_000, + }); + await this.database.workspaceTransaction(event.workspaceId, (tx) => + tx.query( + `UPDATE atlas.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.eventId, this.workerId], + ), + ); + } catch (error) { + await this.database.workspaceTransaction(event.workspaceId, (tx) => + tx.query( + `UPDATE atlas.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', last_error_message = '任务队列暂不可用', updated_at = now() + WHERE id = $1 AND status = 'dispatching' AND locked_by = $2`, + [event.eventId, this.workerId], + ), + ); + throw error; + } + } +} diff --git a/apps/worker/src/ozon-client.ts b/apps/worker/src/ozon-client.ts new file mode 100644 index 0000000..21f9292 --- /dev/null +++ b/apps/worker/src/ozon-client.ts @@ -0,0 +1,46 @@ +/** 401 是认证失败；403 表示密钥权限或账户组合不被允许。 */ +export class OzonCredentialError extends Error { + public constructor(public readonly status: 401 | 403) { + super(`Ozon credential request failed (${status})`); + } +} +/** 认证以外的确定性 4xx，不能误报为凭据无效。 */ +export class OzonRequestError extends Error { + public constructor(public readonly status: number) { + super(`Ozon verification request was rejected (${status})`); + } +} +export class OzonRetryableError extends Error {} + +/** + * Uses the category-tree read endpoint as the credential probe. It has no write side effect and, + * unlike warehouse-list, is verified against the current Ozon account integration. + */ +export async function verifyOzonCredentials( + clientId: string, + apiKey: string, +): Promise { + let response: Response; + try { + response = await fetch('https://api-seller.ozon.ru/v1/description-category/tree', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Client-Id': clientId, + 'Api-Key': apiKey, + }, + body: JSON.stringify({ language: 'ZH_HANS' }), + signal: AbortSignal.timeout(30_000), + }); + } catch { + throw new OzonRetryableError('Ozon request could not be completed'); + } + if (response.ok) return; + if (response.status === 401 || response.status === 403) { + throw new OzonCredentialError(response.status); + } + if (response.status === 429 || response.status >= 500) { + throw new OzonRetryableError(`Ozon temporarily unavailable (${response.status})`); + } + throw new OzonRequestError(response.status); +} diff --git a/apps/worker/src/store-secret.crypto.ts b/apps/worker/src/store-secret.crypto.ts new file mode 100644 index 0000000..41bbee2 --- /dev/null +++ b/apps/worker/src/store-secret.crypto.ts @@ -0,0 +1,20 @@ +import { createDecipheriv, createHash } from 'node:crypto'; + +/** Decrypt only inside a running Worker; callers must never log the returned API key. */ +export function decryptStoreSecret( + ciphertextWithTag: Buffer, + nonce: Buffer, + keyMaterial: string, +): string { + const tagLength = 16; + if (ciphertextWithTag.length <= tagLength) { + throw new Error('Invalid encrypted store secret'); + } + const key = createHash('sha256').update(keyMaterial).digest(); + const decipher = createDecipheriv('aes-256-gcm', key, nonce); + decipher.setAuthTag(ciphertextWithTag.subarray(-tagLength)); + return Buffer.concat([ + decipher.update(ciphertextWithTag.subarray(0, -tagLength)), + decipher.final(), + ]).toString('utf8'); +} diff --git a/docs/api/stores-and-platforms.md b/docs/api/stores-and-platforms.md index 4f05929..4073d6f 100644 --- a/docs/api/stores-and-platforms.md +++ b/docs/api/stores-and-platforms.md @@ -13,6 +13,7 @@ "code": "OZON", "name": "Ozon", "supported_currencies": ["CNY", "RUB"], + "default_currency_code": "CNY", "default_market_code": "RU", "default_locale": "ru-RU" }] @@ -81,7 +82,7 @@ Headers：`Idempotency-Key` 必填。 } ``` -校验 `currency_code` 是否为目标平台可用币种。服务端立即把 API Key 信封加密保存、创建 `status=verifying` 的店铺，随后投递“低风险凭据验证 + 仓库同步”操作；不得将 Key 放入任务 payload、日志或响应。 +校验 `currency_code` 是否为目标平台可用币种。服务端立即把 API Key 信封加密保存、创建 `status=verifying` 的店铺，并投递低风险凭据验证操作；不得将 Key 放入任务 payload、日志或响应。仓库同步是独立的后续操作，不随创建自动触发。 返回 `202 Accepted`： @@ -96,6 +97,32 @@ Headers：`Idempotency-Key` 必填。 首家未删除店铺强制为默认。对已有店铺提交 `is_default=true` 时，服务端在同一事务取消旧默认店铺；校验失败不把店铺置为 `active`，失败原因以可展示的脱敏 `last_error` 摘要（`code`、`message`、`at`）返回给后续读取接口。 +### 4.1 Ozon 凭据验证与前端展示 + +Worker 只调用一个无写副作用的 Ozon Seller API 探针： + +```http +POST /v1/description-category/tree +Client-Id: +Api-Key: +Content-Type: application/json + +{ "language": "ZH_HANS" } +``` + +该调用只用于验证 Ozon 是否接受当前 Client ID 与 API Key，不创建、修改或同步任何 Ozon 数据。它不携带 API Key 至队列、日志、operation 或 HTTP 响应。 + +| Worker 结果 | 店铺最终状态 | 前端文案与行为 | +|---|---|---| +| operation 为 `queued` / `running` | `status=verifying`、`credential_status=unknown` | 仅显示“验证中”；不要同时展示“未验证”。可轮询 operation，并在终态后刷新店铺列表。 | +| Ozon 返回 2xx | `active` / `valid` | 显示“正常”。 | +| Ozon 返回 401 | `credential_invalid` / `invalid` | 显示“凭据无效”，提示用户核对 Client ID 与 API Key。 | +| Ozon 返回 403 | `credential_invalid` / `invalid` | 显示“访问被拒绝”，提示 Client ID 与 API Key 必须属于同一账户，且密钥需要读取权限。 | +| Ozon 返回其他确定性 4xx | `sync_error` / 保留当前凭据状态 | 显示 `last_error.message`；该状态**不代表 API Key 无效**。 | +| 网络超时、429 或 5xx | 保持验证中，Worker 重试；重试耗尽后为 `sync_error` | 显示“暂时无法验证，请稍后重试”。 | + +页面最终以 `GET /api/stores` 返回的店铺状态为准；`GET /api/operations/{operation_id}` 只用于得知异步过程是否结束及读取脱敏错误。点击“重新验证凭据”会重新发起上述一次类目树请求。 + ## 5. 编辑、默认店铺、验证与删除 `PATCH /api/stores/{store_id}` diff --git a/docs/guides/连接数据库.md b/docs/guides/连接数据库.md new file mode 100644 index 0000000..4cfa764 --- /dev/null +++ b/docs/guides/连接数据库.md @@ -0,0 +1,26 @@ + +## 在Cursor中连接数据库 +可以。为这次手工维护 Ozon 币种，请在 SQLTools 用迁移专用账号连接： + +| 字段 | 填写内容 | +|---|---| +| Connection name | `Atlas Local - Migrator` | +| Connection group | 可留空，或填 `Local` | +| Connect using | `Server and Port` | +| Server Address | `localhost` | +| Port | 根目录 `.env` 的 `POSTGRES_PORT`（通常是 `5432`） | +| Database | 根目录 `.env` 的 `POSTGRES_DB`（通常是 `atlas`） | +| Username | `atlas_migrator` | +| Use password | 选择 SQLTools Driver Credentials，并填根目录 `.env` 的 `ATLAS_MIGRATOR_PASSWORD` 值 | + +不要使用 `atlas_app_login`：它给 API/Worker 用，权限不足以执行迁移或平台配置维护。 + +连上后，先执行： + +```sql +SELECT code, name, capabilities +FROM atlas.platforms +WHERE code = 'OZON'; +``` + +确认能读到 Ozon 后，再执行我前一条消息提供的事务 SQL。密码不要贴到聊天里。 \ No newline at end of file diff --git a/docs/project/HANDOFF.md b/docs/project/HANDOFF.md index 36f1130..7ab7f9a 100644 --- a/docs/project/HANDOFF.md +++ b/docs/project/HANDOFF.md @@ -1,12 +1,12 @@ # Atlas Seller Kit 交接说明 -> 更新：2026-09-03。此文是新 AI 对话、架构讨论与开发工作的首读入口。开始开发前，必须先阅读并遵守 [`development-collaboration-contract.md`](./development-collaboration-contract.md)。 +> 更新：2026-09-04。此文是新 AI 对话、架构讨论与开发工作的首读入口。开始开发前，必须先阅读并遵守 [`development-collaboration-contract.md`](./development-collaboration-contract.md)。 ## 1. 项目定位 这是一个全新的云端产品，不是 `ozon-seller-kit` 的本地版重构。目标用户可拥有多个门店，每个门店只能绑定一个平台；平台自身可支持多个市场、国家、语言和币种。因此语言、币种、类目和发布规则不能写死为 Ozon 或中文/俄文。 -当前已完成 pnpm monorepo 初始化：包含 Web、NestJS API、BullMQ Worker、WXT 浏览器扩展、共享 API 契约、数据库包、本地基础设施 Compose 与依赖锁文件。API 已实现健康/readiness、集中配置与安全基础、认证、个人工作空间上下文、平台目录/上下文、门店基础管理和操作查询；真实平台验证与同步 Worker 尚未接入。Web 已具备完整前端框架（布局、路由、主题）、登录/注册页、页头平台选择器与店铺管理页（已接入真实 API），其余业务页面为占位；Worker 与扩展为可运行骨架。不要从旧项目复制 SQLite 模型、无鉴权 API、旧 `/trial/:id` 或商品编辑页组件。可选择性复用已经验证过的纯计价公式、采集扩展思路和 AI 任务经验，但必须先适配多租户和云端边界。 +当前已完成 pnpm monorepo 初始化：包含 Web、NestJS API、BullMQ Worker、WXT 浏览器扩展、共享 API 契约、数据库包、本地基础设施 Compose 与依赖锁文件。API 已实现健康/readiness、集中配置与安全基础、认证、个人工作空间上下文、平台目录/上下文、门店基础管理和操作查询；Worker 已具备首个 Ozon 凭据验证消费链路，仓库同步尚未接入。Web 已具备完整前端框架（布局、路由、主题）、登录/注册页、页头平台选择器与店铺管理页（已接入真实 API），其余业务页面为占位；扩展为可运行骨架。不要从旧项目复制 SQLite 模型、无鉴权 API、旧 `/trial/:id` 或商品编辑页组件。可选择性复用已经验证过的纯计价公式、采集扩展思路和 AI 任务经验，但必须先适配多租户和云端边界。 当前运行时基线：Node 24.20.0、pnpm 11.25.0、PostgreSQL 18.6、Redis 7.4.11、MinIO `RELEASE.2025-09-07T16-13-09Z`；Web 默认端口为 `8100`，API 默认端口为 `8200`，扩展 dev server 默认端口为 `8110`（仅本地开发用），Worker 无 HTTP 端口。完整目录、依赖边界与版本表见 [`architecture/project-structure.md`](../architecture/project-structure.md) 和 [`architecture/technology-and-deployment.md`](../architecture/technology-and-deployment.md)。 @@ -47,6 +47,19 @@ - 修复 API 阻断性 bug（跨领域改动，原因：阻断本功能联调；影响 `apps/api/src/stores/stores.controller.ts`）：tsx/esbuild 不生成 `design:paramtypes` 装饰器元数据，控制器裸类型构造参数在 dev 运行时注入 `undefined`，`GET /api/platforms` 返回 500。已改为与 `auth.controller.ts` 一致的显式 `@Inject(...)`；服务层原本就使用显式 token，不受影响。 - 浏览器实测（全新迁移的本地 PostgreSQL 18 + Redis 7）：注册 → 平台选择器 → 创建/编辑/设默认/删除保护/同步 422 错误路径 → 刷新后会话恢复与列表持久化，全部通过；contracts/api/web typecheck 与 web production build 通过。 +### 1.3 开发进度（2026-09-03 晚）：店铺表单与界面收口 + +- 店铺创建与编辑表单补齐空白字符拦截；API Key 输入上限与冻结契约的 8192 字符保持一致，不在前端自行收紧。 +- 设置页补充取消操作；全局主题目前使用 `borderRadius: 10`、默认控件高度 34，小尺寸控件高度 26，输入类控件单独使用 16px 字号与 40px 高度。按钮保留 Ant Design 默认字号、高度与内边距。 +- 页头面包屑改用 12px `RightOutlined` 分隔符，并完成条目、分隔符与侧栏折叠按钮的垂直对齐；页头水平内边距调整为 `0 20px 0 10px`。店铺与设置页的视觉调整不改变接口或领域行为。 +- Ozon 平台现支持 `CNY` 与 `RUB`；新建 Ozon 店铺时默认预选 `CNY`。默认币种由平台能力 `default_currency_code` 提供，服务端仍以平台支持币种表校验提交值。 + +### 1.4 开发进度（2026-09-04）：凭据验证 Worker + +- Worker 轮询 Transactional Outbox，按工作空间隔离认领事件并投递 BullMQ；任务只携带 store、operation 与 secret ID，绝不携带 API Key。 +- `store_credential_verification` 任务仅在 Worker 内读取当前未撤销密钥、短暂解密并调用 Ozon `POST /v1/description-category/tree`（`{ "language": "ZH_HANS" }`）进行低风险验证。该请求已用真实店铺验证可用；仓库接口留待仓库同步模块按其实际请求契约接入。401 表示认证失败；403 单独提示账户/权限不匹配；其他确定性 4xx 记录脱敏 HTTP 状态并标为验证请求错误，不误报为凭据无效。临时网络、限流和 Ozon 服务端错误最多指数退避重试 5 次。 +- 验证成功回写 `active / valid`；密钥被替换或门店删除时，旧任务取消且不得覆盖新状态。前端在验证中只显示一个标签，避免与“未验证”同时出现。 + ## 2. 已确认产品决策 | 主题 | 决策 | @@ -107,7 +120,7 @@ User → Workspace → Membership / Role 当前优先完成门店纵向闭环，而不是继续扩展商品接口： -1. 实现 Outbox Dispatcher 与 Worker 的首个消费链路：将 `store_credential_verification` 投递到 Worker，解密密钥后调用平台适配器，回写门店/密钥状态；失败信息必须脱敏。 +1. 验证并补齐 Outbox Dispatcher 与凭据验证 Worker 的集成测试：覆盖成功、401/403、限流/超时重试、旧密钥任务取消和无密钥日志。 2. 在凭据验证通过后实现仓库同步 Worker、幂等 upsert 和状态回写，并补充失败重试、死信与集成测试。 3. ~~Web 接入真实认证和门店管理 API~~ 已基本完成：登录/注册、会话刷新、平台上下文、门店列表/创建/编辑/默认切换/删除均已上线（见 1.2）；剩余水印绑定入口待素材/水印模块交付后接入。前端不得保存 API Key 或 refresh token。 4. 接着实现素材直传和水印管理，再开始采集箱与商品编辑；浏览器扩展仅在采集 API 稳定后实现实际解析器。 diff --git a/packages/contracts/src/stores.ts b/packages/contracts/src/stores.ts index 34e7fc0..7297535 100644 --- a/packages/contracts/src/stores.ts +++ b/packages/contracts/src/stores.ts @@ -84,6 +84,8 @@ export type Platform = { supported_currencies: string[]; default_market_code: string | null; default_locale: string | null; + /** 平台新建店铺时预选的计价币种；未配置时前端回退到第一项可用币种。 */ + default_currency_code: string | null; }; /** GET /api/platform-context 返回的页头上下文；没有默认店铺不是错误。 */ diff --git a/packages/database/migrations/005_ozon_cny_default_currency.sql b/packages/database/migrations/005_ozon_cny_default_currency.sql new file mode 100644 index 0000000..6da9f47 --- /dev/null +++ b/packages/database/migrations/005_ozon_cny_default_currency.sql @@ -0,0 +1,15 @@ +-- Ozon 店铺默认以人民币计价；保留 RUB 作为平台支持币种。 +-- 使用独立迁移，确保已初始化的数据库也会获得该配置。 +SET search_path TO atlas, public; + +INSERT INTO platform_currency_support (platform_id, currency_code) +SELECT id, 'CNY'::CHAR(3) +FROM platforms +WHERE code = 'OZON' +ON CONFLICT (platform_id, currency_code) DO UPDATE +SET is_enabled = true; + +UPDATE platforms +SET capabilities = capabilities || jsonb_build_object('default_currency_code', 'CNY'), + updated_at = now() +WHERE code = 'OZON'; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 901a842..b893bc3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -151,6 +151,9 @@ importers: '@atlas/contracts': specifier: workspace:* version: link:../../packages/contracts + '@atlas/database': + specifier: workspace:* + version: link:../../packages/database bullmq: specifier: 6.3.4 version: 6.3.4(ioredis@6.0.0)(pg@8.23.0)