feat(api): 开发平台、门店相关代码
This commit is contained in:
R524809
@@ -0,0 +1,125 @@
|
||||
import { Inject, Injectable } from '@nestjs/common';
|
||||
import type { Database, DatabaseTransaction } from '@atlas/database';
|
||||
import { ApiError } from '../common/api-error.js';
|
||||
import { DATABASE } from '../database/database.module.js';
|
||||
|
||||
type PlatformRow = {
|
||||
id: string;
|
||||
code: string;
|
||||
name: string;
|
||||
capabilities: Record<string, unknown>;
|
||||
supported_currencies: string[];
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class PlatformsService {
|
||||
public constructor(@Inject(DATABASE) private readonly database: Database) {}
|
||||
|
||||
public async list() {
|
||||
const rows = await this.database.query<PlatformRow>(
|
||||
`SELECT p.id, p.code, p.name, p.capabilities,
|
||||
coalesce(array_agg(pcs.currency_code ORDER BY pcs.currency_code)
|
||||
FILTER (WHERE pcs.is_enabled AND c.is_enabled), ARRAY[]::char(3)[]) AS supported_currencies
|
||||
FROM atlas.platforms p
|
||||
LEFT JOIN atlas.platform_currency_support pcs ON pcs.platform_id = p.id
|
||||
LEFT JOIN atlas.currencies c ON c.code = pcs.currency_code
|
||||
WHERE p.status = 'active'
|
||||
GROUP BY p.id
|
||||
ORDER BY CASE WHEN p.code = 'OZON' THEN 0 ELSE 1 END, p.name`,
|
||||
);
|
||||
return rows.rows.map((platform) => ({
|
||||
id: platform.id,
|
||||
code: platform.code,
|
||||
name: platform.name,
|
||||
supported_currencies: platform.supported_currencies,
|
||||
// 能力未配置时使用 null，而不是把 Ozon 的市场或语言假设扩散到通用模型。
|
||||
default_market_code: stringCapability(
|
||||
platform.capabilities,
|
||||
'default_market_code',
|
||||
),
|
||||
default_locale: stringCapability(
|
||||
platform.capabilities,
|
||||
'default_locale',
|
||||
),
|
||||
}));
|
||||
}
|
||||
|
||||
public async context(workspaceId: string, platformId: string) {
|
||||
return this.database.workspaceTransaction(
|
||||
workspaceId,
|
||||
async (transaction) => {
|
||||
const platform = await this.requireActivePlatform(
|
||||
transaction,
|
||||
platformId,
|
||||
);
|
||||
const stores = await transaction.query<{
|
||||
id: string;
|
||||
name: string;
|
||||
status: string;
|
||||
credential_status: string;
|
||||
count: string;
|
||||
}>(
|
||||
`SELECT s.id, s.name, s.status, s.credential_status,
|
||||
count(*) OVER ()::text AS count
|
||||
FROM atlas.stores s
|
||||
WHERE s.platform_id = $1 AND s.deleted_at IS NULL AND s.is_default
|
||||
LIMIT 1`,
|
||||
[platformId],
|
||||
);
|
||||
// 默认门店不存在时仍要正确返回总数，不能用默认行的窗口函数猜测。
|
||||
const count = await transaction.query<{ count: string }>(
|
||||
`SELECT count(*)::text AS count FROM atlas.stores
|
||||
WHERE platform_id = $1 AND deleted_at IS NULL`,
|
||||
[platformId],
|
||||
);
|
||||
const defaultStore = stores.rows[0];
|
||||
return {
|
||||
platform: {
|
||||
id: platform.id,
|
||||
code: platform.code,
|
||||
name: platform.name,
|
||||
},
|
||||
default_store: defaultStore
|
||||
? {
|
||||
id: defaultStore.id,
|
||||
name: defaultStore.name,
|
||||
status: defaultStore.status,
|
||||
credential_status: defaultStore.credential_status,
|
||||
}
|
||||
: null,
|
||||
store_count: Number(count.rows[0]?.count ?? 0),
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private async requireActivePlatform(
|
||||
transaction: DatabaseTransaction,
|
||||
platformId: string,
|
||||
) {
|
||||
const rows = await transaction.query<{
|
||||
id: string;
|
||||
code: string;
|
||||
name: string;
|
||||
}>(
|
||||
`SELECT id, code, name FROM atlas.platforms WHERE id = $1 AND status = 'active'`,
|
||||
[platformId],
|
||||
);
|
||||
if (!rows.rows[0]) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'PLATFORM_UNAVAILABLE',
|
||||
'平台不存在或已停用',
|
||||
);
|
||||
}
|
||||
return rows.rows[0];
|
||||
}
|
||||
}
|
||||
|
||||
function stringCapability(
|
||||
capabilities: Record<string, unknown>,
|
||||
key: string,
|
||||
): string | null {
|
||||
const value = capabilities[key];
|
||||
return typeof value === 'string' ? value : null;
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import {
|
||||
createCipheriv,
|
||||
createHash,
|
||||
createHmac,
|
||||
randomBytes,
|
||||
} from 'node:crypto';
|
||||
|
||||
export type EncryptedStoreSecret = {
|
||||
ciphertext: Buffer;
|
||||
nonce: Buffer;
|
||||
fingerprint: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* 密钥只在此边界短暂以明文存在。数据库保存 AES-GCM 密文和随机 nonce，
|
||||
* fingerprint 是不可逆 HMAC，仅用于安全地识别同一份密钥。
|
||||
*/
|
||||
export function encryptStoreSecret(
|
||||
plaintext: string,
|
||||
keyMaterial: string,
|
||||
): EncryptedStoreSecret {
|
||||
const key = createHash('sha256').update(keyMaterial).digest();
|
||||
const nonce = randomBytes(12);
|
||||
const cipher = createCipheriv('aes-256-gcm', key, nonce);
|
||||
const ciphertext = Buffer.concat([
|
||||
cipher.update(plaintext, 'utf8'),
|
||||
cipher.final(),
|
||||
cipher.getAuthTag(),
|
||||
]);
|
||||
return {
|
||||
ciphertext,
|
||||
nonce,
|
||||
fingerprint: createHmac('sha256', key)
|
||||
.update(plaintext)
|
||||
.digest('base64url'),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
import {
|
||||
Body,
|
||||
Controller,
|
||||
Delete,
|
||||
Get,
|
||||
Headers,
|
||||
HttpCode,
|
||||
Param,
|
||||
Patch,
|
||||
Post,
|
||||
Query,
|
||||
Req,
|
||||
Res,
|
||||
} from '@nestjs/common';
|
||||
import {
|
||||
ApiAcceptedResponse,
|
||||
ApiBearerAuth,
|
||||
ApiNoContentResponse,
|
||||
ApiOkResponse,
|
||||
ApiOperation,
|
||||
ApiTags,
|
||||
} from '@nestjs/swagger';
|
||||
import {
|
||||
createStoreRequestSchema,
|
||||
listStoresQuerySchema,
|
||||
updateStoreRequestSchema,
|
||||
} from '@atlas/contracts';
|
||||
import { z } from 'zod';
|
||||
import { ApiError } from '../common/api-error.js';
|
||||
import type { AuthenticatedRequest } from '../auth/auth.guard.js';
|
||||
import { CurrentWorkspace } from '../workspace/current-workspace.decorator.js';
|
||||
import type { Workspace } from '@atlas/contracts';
|
||||
import { PlatformsService } from './platforms.service.js';
|
||||
import { StoresService } from './stores.service.js';
|
||||
|
||||
type HttpResponse = { status: (status: number) => HttpResponse };
|
||||
|
||||
@ApiTags('Platforms')
|
||||
@ApiBearerAuth('access-token')
|
||||
@Controller()
|
||||
export class PlatformsController {
|
||||
public constructor(private readonly platformsService: PlatformsService) {}
|
||||
|
||||
@Get('platforms')
|
||||
@ApiOperation({ summary: '获取已启用的平台及非敏感能力目录' })
|
||||
@ApiOkResponse({ description: '平台目录' })
|
||||
public async list() {
|
||||
return { data: await this.platformsService.list() };
|
||||
}
|
||||
|
||||
@Get('platform-context')
|
||||
@ApiOperation({ summary: '获取页头所选平台的默认门店上下文' })
|
||||
@ApiOkResponse({ description: '平台、默认门店和门店总数' })
|
||||
public async context(
|
||||
@Headers('x-atlas-platform-id') platformId: string | undefined,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
) {
|
||||
if (!platformId) {
|
||||
throw new ApiError(
|
||||
400,
|
||||
'PLATFORM_CONTEXT_REQUIRED',
|
||||
'请求缺少 X-Atlas-Platform-Id',
|
||||
);
|
||||
}
|
||||
return {
|
||||
data: await this.platformsService.context(
|
||||
workspace.id,
|
||||
requireUuid('X-Atlas-Platform-Id', platformId),
|
||||
),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@ApiTags('Stores')
|
||||
@ApiBearerAuth('access-token')
|
||||
@Controller('stores')
|
||||
export class StoresController {
|
||||
public constructor(private readonly storesService: StoresService) {}
|
||||
|
||||
@Get()
|
||||
@ApiOperation({ summary: '按平台分页列出当前工作空间的门店' })
|
||||
@ApiOkResponse({ description: '门店列表与不透明游标' })
|
||||
public async list(
|
||||
@Query() query: unknown,
|
||||
@Headers('x-atlas-platform-id') platformHeader: string | undefined,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
) {
|
||||
if (platformHeader) {
|
||||
throw new ApiError(
|
||||
400,
|
||||
'PLATFORM_CONTEXT_CONFLICT',
|
||||
'门店管理列表不能同时使用平台 header 和 platform_id 查询参数',
|
||||
);
|
||||
}
|
||||
return this.storesService.list(
|
||||
workspace.id,
|
||||
parse(listStoresQuerySchema, query),
|
||||
);
|
||||
}
|
||||
|
||||
@Get(':storeId')
|
||||
@ApiOperation({ summary: '读取单个门店' })
|
||||
public async get(
|
||||
@Param('storeId') storeId: string,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
) {
|
||||
return {
|
||||
data: await this.storesService.get(
|
||||
workspace.id,
|
||||
requireUuid('store_id', storeId),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
@Post()
|
||||
@ApiOperation({ summary: '创建门店连接，并请求异步凭据验证' })
|
||||
@ApiAcceptedResponse({ description: '已加密保存凭据，并创建验证操作' })
|
||||
public async create(
|
||||
@Body() body: unknown,
|
||||
@Headers('idempotency-key') idempotencyKey: string | undefined,
|
||||
@Req() request: AuthenticatedRequest,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
@Res({ passthrough: true }) response: HttpResponse,
|
||||
) {
|
||||
return withStatus(
|
||||
response,
|
||||
await this.storesService.create(
|
||||
workspace.id,
|
||||
request.auth.userId,
|
||||
requireIdempotencyKey(idempotencyKey),
|
||||
parse(createStoreRequestSchema, body),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@Patch(':storeId')
|
||||
@ApiOperation({ summary: '按版本编辑门店；更新密钥会重新请求验证' })
|
||||
@ApiOkResponse({ description: '普通字段已更新' })
|
||||
@ApiAcceptedResponse({ description: '新密钥已保存，并创建验证操作' })
|
||||
public async update(
|
||||
@Param('storeId') storeId: string,
|
||||
@Body() body: unknown,
|
||||
@Headers('idempotency-key') idempotencyKey: string | undefined,
|
||||
@Headers('if-match') ifMatch: string | undefined,
|
||||
@Req() request: AuthenticatedRequest,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
@Res({ passthrough: true }) response: HttpResponse,
|
||||
) {
|
||||
return withStatus(
|
||||
response,
|
||||
await this.storesService.update(
|
||||
workspace.id,
|
||||
request.auth.userId,
|
||||
requireUuid('store_id', storeId),
|
||||
requireIdempotencyKey(idempotencyKey),
|
||||
parseRevision(ifMatch),
|
||||
parse(updateStoreRequestSchema, body),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@Post(':storeId/set-default')
|
||||
@ApiOperation({ summary: '设为当前平台默认门店' })
|
||||
public async setDefault(
|
||||
@Param('storeId') storeId: string,
|
||||
@Headers('idempotency-key') idempotencyKey: string | undefined,
|
||||
@Req() request: AuthenticatedRequest,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
@Res({ passthrough: true }) response: HttpResponse,
|
||||
) {
|
||||
return withStatus(
|
||||
response,
|
||||
await this.storesService.setDefault(
|
||||
workspace.id,
|
||||
request.auth.userId,
|
||||
requireUuid('store_id', storeId),
|
||||
requireIdempotencyKey(idempotencyKey),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@Post(':storeId/credential-verifications')
|
||||
@ApiOperation({ summary: '手动请求重新验证门店凭据' })
|
||||
@ApiAcceptedResponse({ description: '凭据验证操作已入队' })
|
||||
public async credentialVerification(
|
||||
@Param('storeId') storeId: string,
|
||||
@Headers('idempotency-key') idempotencyKey: string | undefined,
|
||||
@Req() request: AuthenticatedRequest,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
@Res({ passthrough: true }) response: HttpResponse,
|
||||
) {
|
||||
return withStatus(
|
||||
response,
|
||||
await this.storesService.requestCredentialVerification(
|
||||
workspace.id,
|
||||
request.auth.userId,
|
||||
requireUuid('store_id', storeId),
|
||||
requireIdempotencyKey(idempotencyKey),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@Post(':storeId/warehouse-syncs')
|
||||
@ApiOperation({ summary: '请求异步同步门店仓库' })
|
||||
@ApiAcceptedResponse({ description: '仓库同步操作已入队' })
|
||||
public async warehouseSync(
|
||||
@Param('storeId') storeId: string,
|
||||
@Headers('idempotency-key') idempotencyKey: string | undefined,
|
||||
@Req() request: AuthenticatedRequest,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
@Res({ passthrough: true }) response: HttpResponse,
|
||||
) {
|
||||
return withStatus(
|
||||
response,
|
||||
await this.storesService.requestWarehouseSync(
|
||||
workspace.id,
|
||||
request.auth.userId,
|
||||
requireUuid('store_id', storeId),
|
||||
requireIdempotencyKey(idempotencyKey),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@Get(':storeId/warehouses')
|
||||
@ApiOperation({ summary: '读取最近成功同步的仓库摘要' })
|
||||
public async warehouses(
|
||||
@Param('storeId') storeId: string,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
) {
|
||||
return this.storesService.warehouses(
|
||||
workspace.id,
|
||||
requireUuid('store_id', storeId),
|
||||
);
|
||||
}
|
||||
|
||||
@Delete(':storeId')
|
||||
@HttpCode(204)
|
||||
@ApiOperation({ summary: '软删除门店连接并撤销当前密钥' })
|
||||
@ApiNoContentResponse({ description: '门店已删除' })
|
||||
public async remove(
|
||||
@Param('storeId') storeId: string,
|
||||
@Headers('idempotency-key') idempotencyKey: string | undefined,
|
||||
@Req() request: AuthenticatedRequest,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
@Res({ passthrough: true }) response: HttpResponse,
|
||||
) {
|
||||
return withStatus(
|
||||
response,
|
||||
await this.storesService.remove(
|
||||
workspace.id,
|
||||
request.auth.userId,
|
||||
requireUuid('store_id', storeId),
|
||||
requireIdempotencyKey(idempotencyKey),
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@ApiTags('Operations')
|
||||
@ApiBearerAuth('access-token')
|
||||
@Controller('operations')
|
||||
export class OperationsController {
|
||||
public constructor(private readonly storesService: StoresService) {}
|
||||
|
||||
@Get(':operationId')
|
||||
@ApiOperation({ summary: '读取当前工作空间的异步操作状态' })
|
||||
public async get(
|
||||
@Param('operationId') operationId: string,
|
||||
@CurrentWorkspace() workspace: Workspace,
|
||||
) {
|
||||
return {
|
||||
data: await this.storesService.operation(
|
||||
workspace.id,
|
||||
requireUuid('operation_id', operationId),
|
||||
),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
function withStatus(
|
||||
response: HttpResponse,
|
||||
result: { status: number; body: unknown },
|
||||
) {
|
||||
response.status(result.status);
|
||||
return result.body;
|
||||
}
|
||||
|
||||
function requireIdempotencyKey(value: string | undefined): string {
|
||||
if (!value) {
|
||||
throw new ApiError(
|
||||
400,
|
||||
'IDEMPOTENCY_KEY_REQUIRED',
|
||||
'请求缺少 Idempotency-Key',
|
||||
);
|
||||
}
|
||||
return requireUuid('Idempotency-Key', value);
|
||||
}
|
||||
|
||||
function requireUuid(field: string, value: string): string {
|
||||
if (z.uuid().safeParse(value).success) return value;
|
||||
throw new ApiError(422, 'VALIDATION_FAILED', '请求参数不符合要求', [
|
||||
{ field, code: 'INVALID_FIELD', message: `${field} 必须是 UUID` },
|
||||
]);
|
||||
}
|
||||
|
||||
function parseRevision(value: string | undefined): number {
|
||||
const match = value?.match(/^"([1-9]\d*)"$/);
|
||||
if (!match) {
|
||||
throw new ApiError(
|
||||
400,
|
||||
'IF_MATCH_REQUIRED',
|
||||
'请求必须携带形如 "1" 的 If-Match',
|
||||
);
|
||||
}
|
||||
return Number(match[1]);
|
||||
}
|
||||
|
||||
function parse<T>(schema: z.ZodType<T>, value: unknown): T {
|
||||
const result = schema.safeParse(value);
|
||||
if (result.success) return result.data;
|
||||
throw new ApiError(
|
||||
422,
|
||||
'VALIDATION_FAILED',
|
||||
'请求参数不符合要求',
|
||||
result.error.issues.map((issue) => ({
|
||||
field: issue.path.join('.'),
|
||||
code: 'INVALID_FIELD',
|
||||
message: issue.message,
|
||||
})),
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { IdempotencyService } from '../common/idempotency.service.js';
|
||||
import {
|
||||
OperationsController,
|
||||
PlatformsController,
|
||||
StoresController,
|
||||
} from './stores.controller.js';
|
||||
import { PlatformsService } from './platforms.service.js';
|
||||
import { StoresService } from './stores.service.js';
|
||||
|
||||
@Module({
|
||||
controllers: [PlatformsController, StoresController, OperationsController],
|
||||
providers: [IdempotencyService, PlatformsService, StoresService],
|
||||
})
|
||||
export class StoresModule {}
|
||||
@@ -0,0 +1,729 @@
|
||||
import { Inject, Injectable } from '@nestjs/common';
|
||||
import type {
|
||||
CreateStoreRequest,
|
||||
ListStoresQuery,
|
||||
Operation,
|
||||
Store,
|
||||
UpdateStoreRequest,
|
||||
} from '@atlas/contracts';
|
||||
import type { Database, DatabaseTransaction } from '@atlas/database';
|
||||
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';
|
||||
import { encryptStoreSecret } from './store-secret.crypto.js';
|
||||
|
||||
type StoreRow = Omit<Store, keyof { created_at: never; updated_at: never }> & {
|
||||
credential_checked_at: Date | string | null;
|
||||
api_key_expires_at: Date | string | null;
|
||||
product_limit_checked_at: Date | string | null;
|
||||
warehouses_synced_at: Date | string | null;
|
||||
created_at: Date | string;
|
||||
updated_at: Date | string;
|
||||
};
|
||||
type StoreLockRow = {
|
||||
id: string;
|
||||
platform_id: string;
|
||||
revision: number;
|
||||
is_default: boolean;
|
||||
status: string;
|
||||
};
|
||||
type OperationRow = {
|
||||
id: string;
|
||||
kind: string;
|
||||
status: string;
|
||||
created_at: Date | string;
|
||||
completed_at: Date | string | null;
|
||||
error_code: string | null;
|
||||
error_message: string | null;
|
||||
};
|
||||
type WarehouseRow = {
|
||||
id: string;
|
||||
platform_warehouse_id: string;
|
||||
name: string;
|
||||
fulfillment: string;
|
||||
is_rfbs: boolean | null;
|
||||
status: string;
|
||||
synced_at: Date | string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class StoresService {
|
||||
public constructor(
|
||||
private readonly idempotency: IdempotencyService,
|
||||
@Inject(DATABASE) private readonly database: Database,
|
||||
@Inject(APP_CONFIG) private readonly config: AppConfig,
|
||||
) {}
|
||||
|
||||
public async list(workspaceId: string, query: ListStoresQuery) {
|
||||
const cursor = query.cursor ? decodeCursor(query.cursor) : null;
|
||||
return this.database.workspaceTransaction(
|
||||
workspaceId,
|
||||
async (transaction: DatabaseTransaction) => {
|
||||
const rows = await transaction.query<StoreRow>(
|
||||
`${storeSelect}
|
||||
WHERE s.platform_id = $1 AND s.deleted_at IS NULL
|
||||
AND ($2::timestamptz IS NULL OR (s.created_at, s.id) < ($2::timestamptz, $3::uuid))
|
||||
GROUP BY s.id, binding.value
|
||||
ORDER BY s.created_at DESC, s.id DESC
|
||||
LIMIT $4`,
|
||||
[
|
||||
query.platform_id,
|
||||
cursor?.createdAt ?? null,
|
||||
cursor?.id ?? null,
|
||||
query.limit + 1,
|
||||
],
|
||||
);
|
||||
const hasMore = rows.rows.length > query.limit;
|
||||
const items = rows.rows.slice(0, query.limit).map(toStore);
|
||||
const last = items.at(-1);
|
||||
return {
|
||||
data: items,
|
||||
meta: {
|
||||
next_cursor:
|
||||
hasMore && last
|
||||
? encodeCursor(last.created_at, last.id)
|
||||
: null,
|
||||
has_more: hasMore,
|
||||
},
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async get(workspaceId: string, storeId: string): Promise<Store> {
|
||||
return this.database.workspaceTransaction(
|
||||
workspaceId,
|
||||
async (transaction: DatabaseTransaction) =>
|
||||
this.requireStore(transaction, storeId),
|
||||
);
|
||||
}
|
||||
|
||||
public async create(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
idempotencyKey: string,
|
||||
input: CreateStoreRequest,
|
||||
) {
|
||||
return this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'POST',
|
||||
path: '/api/stores',
|
||||
key: idempotencyKey,
|
||||
request: input,
|
||||
},
|
||||
async (transaction) => {
|
||||
await this.assertPlatformCurrency(
|
||||
transaction,
|
||||
input.platform_id,
|
||||
input.currency_code,
|
||||
);
|
||||
// 同一平台的默认店铺修改串行化，避免两个并发创建都错误地判断自己是首店。
|
||||
await lockPlatform(transaction, workspaceId, input.platform_id);
|
||||
const existing = await transaction.query<{ id: string }>(
|
||||
`SELECT id FROM atlas.stores
|
||||
WHERE platform_id = $1 AND platform_client_id = $2
|
||||
LIMIT 1`,
|
||||
[input.platform_id, input.client_id],
|
||||
);
|
||||
if (existing.rows[0]) {
|
||||
throw new ApiError(
|
||||
409,
|
||||
'STORE_CONNECTION_ALREADY_EXISTS',
|
||||
'该平台账户已绑定为门店',
|
||||
);
|
||||
}
|
||||
const currentStores = await transaction.query<{
|
||||
count: string;
|
||||
}>(
|
||||
`SELECT count(*)::text AS count FROM atlas.stores
|
||||
WHERE platform_id = $1 AND deleted_at IS NULL`,
|
||||
[input.platform_id],
|
||||
);
|
||||
const becomesDefault =
|
||||
Number(currentStores.rows[0]?.count ?? 0) === 0 ||
|
||||
input.is_default;
|
||||
if (becomesDefault) {
|
||||
await transaction.query(
|
||||
`UPDATE atlas.stores SET is_default = false, revision = revision + 1, updated_at = now()
|
||||
WHERE platform_id = $1 AND is_default AND deleted_at IS NULL`,
|
||||
[input.platform_id],
|
||||
);
|
||||
}
|
||||
const created = await transaction.query<{ id: string }>(
|
||||
`INSERT INTO atlas.stores
|
||||
(workspace_id, platform_id, name, platform_client_id, currency_code, is_default, note)
|
||||
VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, $4, $5, $6)
|
||||
RETURNING id`,
|
||||
[
|
||||
input.platform_id,
|
||||
input.name,
|
||||
input.client_id,
|
||||
input.currency_code,
|
||||
becomesDefault,
|
||||
input.note ?? null,
|
||||
],
|
||||
);
|
||||
const storeId = created.rows[0].id;
|
||||
await this.replaceApiKey(transaction, storeId, input.api_key);
|
||||
const operation = await this.createOperation(
|
||||
transaction,
|
||||
userId,
|
||||
storeId,
|
||||
'store_credential_verification',
|
||||
);
|
||||
return {
|
||||
status: 202,
|
||||
body: {
|
||||
data: {
|
||||
store: await this.requireStore(
|
||||
transaction,
|
||||
storeId,
|
||||
),
|
||||
operation,
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async update(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
idempotencyKey: string,
|
||||
revision: number,
|
||||
input: UpdateStoreRequest,
|
||||
) {
|
||||
return this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'PATCH',
|
||||
path: `/api/stores/${storeId}`,
|
||||
key: idempotencyKey,
|
||||
request: { if_match: revision, ...input },
|
||||
},
|
||||
async (transaction) => {
|
||||
const current = await this.lockStore(transaction, storeId);
|
||||
this.assertRevision(current, revision);
|
||||
if (input.currency_code) {
|
||||
await this.assertPlatformCurrency(
|
||||
transaction,
|
||||
current.platform_id,
|
||||
input.currency_code,
|
||||
);
|
||||
}
|
||||
const replacingKey = input.api_key !== undefined;
|
||||
await transaction.query(
|
||||
`UPDATE atlas.stores
|
||||
SET name = COALESCE($1, name), currency_code = COALESCE($2, currency_code),
|
||||
note = CASE WHEN $3 THEN $4 ELSE note END,
|
||||
status = CASE WHEN $5 THEN 'verifying'::atlas.store_status ELSE status END,
|
||||
credential_status = CASE WHEN $5 THEN 'unknown'::atlas.credential_status ELSE credential_status END,
|
||||
last_error_code = CASE WHEN $5 THEN NULL ELSE last_error_code END,
|
||||
last_error_message = CASE WHEN $5 THEN NULL ELSE last_error_message END,
|
||||
last_error_at = CASE WHEN $5 THEN NULL ELSE last_error_at END,
|
||||
revision = revision + 1, updated_at = now()
|
||||
WHERE id = $6`,
|
||||
[
|
||||
input.name ?? null,
|
||||
input.currency_code ?? null,
|
||||
Object.hasOwn(input, 'note'),
|
||||
input.note ?? null,
|
||||
replacingKey,
|
||||
storeId,
|
||||
],
|
||||
);
|
||||
if (input.api_key !== undefined) {
|
||||
await this.replaceApiKey(
|
||||
transaction,
|
||||
storeId,
|
||||
input.api_key,
|
||||
);
|
||||
const operation = await this.createOperation(
|
||||
transaction,
|
||||
userId,
|
||||
storeId,
|
||||
'store_credential_verification',
|
||||
);
|
||||
return {
|
||||
status: 202,
|
||||
body: {
|
||||
data: {
|
||||
store: await this.requireStore(
|
||||
transaction,
|
||||
storeId,
|
||||
),
|
||||
operation,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
return {
|
||||
status: 200,
|
||||
body: {
|
||||
data: await this.requireStore(transaction, storeId),
|
||||
},
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async setDefault(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
idempotencyKey: string,
|
||||
) {
|
||||
return this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'POST',
|
||||
path: `/api/stores/${storeId}/set-default`,
|
||||
key: idempotencyKey,
|
||||
request: {},
|
||||
},
|
||||
async (transaction) => {
|
||||
const current = await this.lockStore(transaction, storeId);
|
||||
await lockPlatform(
|
||||
transaction,
|
||||
workspaceId,
|
||||
current.platform_id,
|
||||
);
|
||||
await transaction.query(
|
||||
`UPDATE atlas.stores SET is_default = false, revision = revision + 1, updated_at = now()
|
||||
WHERE platform_id = $1 AND is_default AND deleted_at IS NULL AND id <> $2`,
|
||||
[current.platform_id, storeId],
|
||||
);
|
||||
if (!current.is_default) {
|
||||
await transaction.query(
|
||||
`UPDATE atlas.stores SET is_default = true, revision = revision + 1, updated_at = now()
|
||||
WHERE id = $1`,
|
||||
[storeId],
|
||||
);
|
||||
}
|
||||
return {
|
||||
status: 200,
|
||||
body: {
|
||||
data: await this.requireStore(transaction, storeId),
|
||||
},
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async requestCredentialVerification(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
idempotencyKey: string,
|
||||
) {
|
||||
return this.requestOperation(
|
||||
workspaceId,
|
||||
userId,
|
||||
storeId,
|
||||
idempotencyKey,
|
||||
'credential-verifications',
|
||||
'store_credential_verification',
|
||||
false,
|
||||
);
|
||||
}
|
||||
|
||||
public async requestWarehouseSync(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
idempotencyKey: string,
|
||||
) {
|
||||
return this.requestOperation(
|
||||
workspaceId,
|
||||
userId,
|
||||
storeId,
|
||||
idempotencyKey,
|
||||
'warehouse-syncs',
|
||||
'store_warehouse_sync',
|
||||
true,
|
||||
);
|
||||
}
|
||||
|
||||
public async remove(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
idempotencyKey: string,
|
||||
) {
|
||||
return this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'DELETE',
|
||||
path: `/api/stores/${storeId}`,
|
||||
key: idempotencyKey,
|
||||
request: {},
|
||||
},
|
||||
async (transaction) => {
|
||||
const current = await this.lockStore(transaction, storeId);
|
||||
if (current.is_default) {
|
||||
const replacement = await transaction.query<{ id: string }>(
|
||||
`SELECT id FROM atlas.stores
|
||||
WHERE platform_id = $1 AND id <> $2 AND deleted_at IS NULL LIMIT 1`,
|
||||
[current.platform_id, storeId],
|
||||
);
|
||||
if (replacement.rows[0]) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'DEFAULT_STORE_REPLACEMENT_REQUIRED',
|
||||
'删除默认门店前必须先选择同平台的其他默认门店',
|
||||
);
|
||||
}
|
||||
}
|
||||
await transaction.query(
|
||||
`UPDATE atlas.stores
|
||||
SET deleted_at = now(), status = 'deleted', is_default = false,
|
||||
revision = revision + 1, updated_at = now()
|
||||
WHERE id = $1`,
|
||||
[storeId],
|
||||
);
|
||||
// 删除连接后立刻撤销可用密钥，避免后台任务或未来恢复流程继续使用它。
|
||||
await transaction.query(
|
||||
`UPDATE atlas.store_secrets SET revoked_at = now(), updated_at = now()
|
||||
WHERE store_id = $1 AND revoked_at IS NULL`,
|
||||
[storeId],
|
||||
);
|
||||
return { status: 204, body: null };
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async warehouses(workspaceId: string, storeId: string) {
|
||||
return this.database.workspaceTransaction(
|
||||
workspaceId,
|
||||
async (transaction: DatabaseTransaction) => {
|
||||
const store = await this.requireStore(transaction, storeId);
|
||||
const rows = await transaction.query<WarehouseRow>(
|
||||
`SELECT id, platform_warehouse_id, name, fulfillment, is_rfbs, status, synced_at
|
||||
FROM atlas.store_warehouses WHERE store_id = $1 ORDER BY name, id`,
|
||||
[storeId],
|
||||
);
|
||||
return {
|
||||
data: rows.rows.map((row) => ({
|
||||
...row,
|
||||
synced_at: toIso(row.synced_at),
|
||||
})),
|
||||
meta: { warehouses_synced_at: store.warehouses_synced_at },
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async operation(workspaceId: string, operationId: string) {
|
||||
return this.database.workspaceTransaction(
|
||||
workspaceId,
|
||||
async (transaction: DatabaseTransaction) => {
|
||||
const rows = await transaction.query<OperationRow>(
|
||||
`SELECT id, kind, status, created_at, completed_at, error_code, error_message
|
||||
FROM atlas.operations WHERE id = $1`,
|
||||
[operationId],
|
||||
);
|
||||
if (!rows.rows[0]) throw notFound('操作');
|
||||
return toOperation(rows.rows[0]);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private async requestOperation(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
idempotencyKey: string,
|
||||
pathSuffix: string,
|
||||
kind: string,
|
||||
requiresValidCredential: boolean,
|
||||
) {
|
||||
return this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'POST',
|
||||
path: `/api/stores/${storeId}/${pathSuffix}`,
|
||||
key: idempotencyKey,
|
||||
request: {},
|
||||
},
|
||||
async (transaction) => {
|
||||
const store = await this.requireStore(transaction, storeId);
|
||||
if (store.status === 'disabled') {
|
||||
throw new ApiError(422, 'STORE_DISABLED', '门店已停用');
|
||||
}
|
||||
if (
|
||||
requiresValidCredential &&
|
||||
store.credential_status !== 'valid'
|
||||
) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'STORE_CREDENTIAL_UNAVAILABLE',
|
||||
'门店凭据尚未验证通过，无法同步仓库',
|
||||
);
|
||||
}
|
||||
const operation = await this.createOperation(
|
||||
transaction,
|
||||
userId,
|
||||
storeId,
|
||||
kind,
|
||||
);
|
||||
return { status: 202, body: { data: { operation } } };
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private async assertPlatformCurrency(
|
||||
transaction: DatabaseTransaction,
|
||||
platformId: string,
|
||||
currencyCode: string,
|
||||
) {
|
||||
const platform = await transaction.query<{ id: string }>(
|
||||
`SELECT id FROM atlas.platforms WHERE id = $1 AND status = 'active'`,
|
||||
[platformId],
|
||||
);
|
||||
if (!platform.rows[0]) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'PLATFORM_UNAVAILABLE',
|
||||
'平台不存在或已停用',
|
||||
);
|
||||
}
|
||||
const supported = await transaction.query<{ currency_code: string }>(
|
||||
`SELECT pcs.currency_code FROM atlas.platform_currency_support pcs
|
||||
JOIN atlas.currencies c ON c.code = pcs.currency_code
|
||||
WHERE pcs.platform_id = $1 AND pcs.currency_code = $2
|
||||
AND pcs.is_enabled AND c.is_enabled`,
|
||||
[platformId, currencyCode],
|
||||
);
|
||||
if (!supported.rows[0]) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'PLATFORM_CURRENCY_UNSUPPORTED',
|
||||
'该平台不支持此默认计价币种',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async lockStore(transaction: DatabaseTransaction, storeId: string) {
|
||||
const rows = await transaction.query<StoreLockRow>(
|
||||
`SELECT id, platform_id, revision, is_default, status
|
||||
FROM atlas.stores WHERE id = $1 AND deleted_at IS NULL FOR UPDATE`,
|
||||
[storeId],
|
||||
);
|
||||
if (!rows.rows[0]) throw notFound('门店');
|
||||
return rows.rows[0];
|
||||
}
|
||||
|
||||
private assertRevision(current: StoreLockRow, expected: number) {
|
||||
if (current.revision !== expected) {
|
||||
throw new ApiError(
|
||||
409,
|
||||
'VERSION_CONFLICT',
|
||||
'门店已被其他操作更新',
|
||||
[
|
||||
{
|
||||
field: 'If-Match',
|
||||
code: 'VERSION_CONFLICT',
|
||||
message: `当前版本为 ${current.revision}`,
|
||||
current_revision: current.revision,
|
||||
},
|
||||
],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async replaceApiKey(
|
||||
transaction: DatabaseTransaction,
|
||||
storeId: string,
|
||||
apiKey: string,
|
||||
) {
|
||||
const secret = encryptStoreSecret(
|
||||
apiKey,
|
||||
this.config.STORE_SECRET_ENCRYPTION_KEY,
|
||||
);
|
||||
const existing = await transaction.query<{ id: string }>(
|
||||
`SELECT id FROM atlas.store_secrets
|
||||
WHERE store_id = $1 AND kind = 'seller_api_key' AND fingerprint = $2
|
||||
FOR UPDATE`,
|
||||
[storeId, secret.fingerprint],
|
||||
);
|
||||
await transaction.query(
|
||||
`UPDATE atlas.store_secrets SET revoked_at = now(), updated_at = now()
|
||||
WHERE store_id = $1 AND kind = 'seller_api_key' AND revoked_at IS NULL`,
|
||||
[storeId],
|
||||
);
|
||||
if (existing.rows[0]) {
|
||||
// 同一密钥可被重新提交用于验证；恢复历史行避免违反防重指纹唯一约束。
|
||||
await transaction.query(
|
||||
`UPDATE atlas.store_secrets
|
||||
SET secret_ciphertext = $1, secret_nonce = $2, encryption_key_version = $3,
|
||||
status = 'unknown', revoked_at = NULL, updated_at = now()
|
||||
WHERE id = $4`,
|
||||
[
|
||||
secret.ciphertext,
|
||||
secret.nonce,
|
||||
this.config.STORE_SECRET_ENCRYPTION_KEY_VERSION,
|
||||
existing.rows[0].id,
|
||||
],
|
||||
);
|
||||
return;
|
||||
}
|
||||
await transaction.query(
|
||||
`INSERT INTO atlas.store_secrets
|
||||
(workspace_id, store_id, kind, secret_ciphertext, secret_nonce, encryption_key_version, fingerprint)
|
||||
VALUES (current_setting('app.workspace_id')::uuid, $1, 'seller_api_key', $2, $3, $4, $5)`,
|
||||
[
|
||||
storeId,
|
||||
secret.ciphertext,
|
||||
secret.nonce,
|
||||
this.config.STORE_SECRET_ENCRYPTION_KEY_VERSION,
|
||||
secret.fingerprint,
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
private async createOperation(
|
||||
transaction: DatabaseTransaction,
|
||||
userId: string,
|
||||
storeId: string,
|
||||
kind: string,
|
||||
): Promise<Operation> {
|
||||
const created = await transaction.query<OperationRow>(
|
||||
`INSERT INTO atlas.operations
|
||||
(workspace_id, kind, resource_type, resource_id, created_by_user_id)
|
||||
VALUES (current_setting('app.workspace_id')::uuid, $1, 'store', $2, $3)
|
||||
RETURNING id, kind, status, created_at, completed_at, error_code, error_message`,
|
||||
[kind, storeId, userId],
|
||||
);
|
||||
const operation = toOperation(created.rows[0]);
|
||||
await transaction.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, 'store', $1, $2, $3, $4)`,
|
||||
[
|
||||
storeId,
|
||||
`${kind}.requested`,
|
||||
{ store_id: storeId, operation_id: operation.id },
|
||||
operation.id,
|
||||
],
|
||||
);
|
||||
return operation;
|
||||
}
|
||||
|
||||
private async requireStore(
|
||||
transaction: DatabaseTransaction,
|
||||
storeId: string,
|
||||
): Promise<Store> {
|
||||
const rows = await transaction.query<StoreRow>(
|
||||
`${storeSelect}
|
||||
WHERE s.id = $1 AND s.deleted_at IS NULL
|
||||
GROUP BY s.id, binding.value`,
|
||||
[storeId],
|
||||
);
|
||||
if (!rows.rows[0]) throw notFound('门店');
|
||||
return toStore(rows.rows[0]);
|
||||
}
|
||||
}
|
||||
|
||||
const storeSelect = `
|
||||
SELECT s.id, s.platform_id, s.name, s.platform_client_id, s.currency_code,
|
||||
s.is_default, s.note, s.status, s.credential_status, s.credential_checked_at,
|
||||
CASE WHEN s.last_error_code IS NULL THEN NULL
|
||||
ELSE jsonb_build_object('code', s.last_error_code, 'message', s.last_error_message, 'at', s.last_error_at)
|
||||
END AS last_error,
|
||||
s.api_key_expires_at, s.product_limit, s.product_limit_checked_at,
|
||||
count(sw.id)::integer AS warehouse_count, max(sw.synced_at) AS warehouses_synced_at,
|
||||
binding.value AS watermark_template, s.created_at, s.updated_at, s.revision
|
||||
FROM atlas.stores s
|
||||
LEFT JOIN atlas.store_warehouses sw ON sw.store_id = s.id
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT jsonb_build_object('id', wt.id, 'name', wt.name) AS value
|
||||
FROM atlas.store_watermark_bindings wb
|
||||
JOIN atlas.watermark_templates wt ON wt.id = wb.template_id
|
||||
WHERE wb.store_id = s.id AND wb.unbound_at IS NULL
|
||||
LIMIT 1
|
||||
) binding ON true`;
|
||||
|
||||
function toStore(row: StoreRow): Store {
|
||||
return {
|
||||
...row,
|
||||
credential_checked_at: toIso(row.credential_checked_at),
|
||||
api_key_expires_at: toIso(row.api_key_expires_at),
|
||||
product_limit_checked_at: toIso(row.product_limit_checked_at),
|
||||
warehouses_synced_at: toIso(row.warehouses_synced_at),
|
||||
created_at: toRequiredIso(row.created_at),
|
||||
updated_at: toRequiredIso(row.updated_at),
|
||||
cookie_status: null,
|
||||
};
|
||||
}
|
||||
|
||||
function toOperation(row: OperationRow): Operation {
|
||||
return {
|
||||
id: row.id,
|
||||
kind: row.kind,
|
||||
status: row.status,
|
||||
created_at: toRequiredIso(row.created_at),
|
||||
completed_at: toIso(row.completed_at),
|
||||
error:
|
||||
row.error_code && row.error_message
|
||||
? { code: row.error_code, message: row.error_message }
|
||||
: null,
|
||||
};
|
||||
}
|
||||
|
||||
function toIso(value: Date | string | null): string | null {
|
||||
if (value === null) return null;
|
||||
return value instanceof Date ? value.toISOString() : value;
|
||||
}
|
||||
|
||||
function toRequiredIso(value: Date | string): string {
|
||||
return value instanceof Date ? value.toISOString() : value;
|
||||
}
|
||||
|
||||
function notFound(resource: string) {
|
||||
return new ApiError(404, 'RESOURCE_NOT_FOUND', `${resource}不存在`);
|
||||
}
|
||||
|
||||
async function lockPlatform(
|
||||
transaction: DatabaseTransaction,
|
||||
workspaceId: string,
|
||||
platformId: string,
|
||||
) {
|
||||
await transaction.query(
|
||||
`SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`,
|
||||
[`${workspaceId}:${platformId}`],
|
||||
);
|
||||
}
|
||||
|
||||
function encodeCursor(createdAt: string, id: string): string {
|
||||
return Buffer.from(JSON.stringify({ createdAt, id })).toString('base64url');
|
||||
}
|
||||
|
||||
function decodeCursor(cursor: string): { createdAt: string; id: string } {
|
||||
try {
|
||||
const value = JSON.parse(
|
||||
Buffer.from(cursor, 'base64url').toString('utf8'),
|
||||
) as { createdAt?: unknown; id?: unknown };
|
||||
if (
|
||||
typeof value.createdAt !== 'string' ||
|
||||
Number.isNaN(Date.parse(value.createdAt)) ||
|
||||
typeof value.id !== 'string' ||
|
||||
!/^[0-9a-f]{8}-[0-9a-f-]{27}$/i.test(value.id)
|
||||
) {
|
||||
throw new Error('invalid cursor');
|
||||
}
|
||||
return { createdAt: value.createdAt, id: value.id };
|
||||
} catch {
|
||||
throw new ApiError(422, 'VALIDATION_FAILED', '分页游标无效', [
|
||||
{ field: 'cursor', code: 'INVALID_FIELD', message: '分页游标无效' },
|
||||
]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user