feat(store): 添加门店、门店验证
This commit is contained in:
R524809
+192
-18
@@ -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<OutboxJob>('atlas-dispatch', { connection });
|
||||
const dispatcher = new OutboxDispatcher(database, queue, `worker_${randomUUID()}`);
|
||||
|
||||
const worker = new Worker<OutboxJob>(
|
||||
'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<typeof setInterval> | 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());
|
||||
|
||||
Reference in New Issue
Block a user