import assert from 'node:assert/strict'; import { randomUUID } from 'node:crypto'; import type { AddressInfo } from 'node:net'; import type { Database } from '@atlas/database'; import type { NestExpressApplication } from '@nestjs/platform-express'; import { afterAll, beforeAll, describe, it } from 'vitest'; import { DATABASE } from '../src/database/database.module.js'; import { createApplication } from '../src/main.js'; import { OutboxDispatcher } from '../../worker/src/outbox-dispatcher.js'; import { flattenCategories } from '../../worker/src/ozon-catalog-sync.js'; import { scheduleStoreCategoryTreeSync } from '../../worker/src/store-category-tree-scheduler.js'; let app: NestExpressApplication; let database: Database; let baseUrl = ''; let workspaceId = ''; let userId = ''; let token = ''; let platformId = ''; let storeId = ''; let secretId = ''; async function request(path: string, init: RequestInit = {}) { const response = await fetch(`${baseUrl}${path}`, { ...init, headers: { 'Content-Type': 'application/json', ...init.headers }, }); return { response, body: (await response.json()) as any }; } async function scheduleWithStoreLock() { return database.workspaceTransaction(workspaceId, async (transaction) => { await transaction.query( 'SELECT id FROM atlas.stores WHERE id = $1 FOR UPDATE', [storeId], ); return scheduleStoreCategoryTreeSync(transaction, { storeId, platformId, secretId, }); }); } describe('门店类目树 Worker 调度集成测试', () => { it('类型叶子继承父类目 ID,并使用 type_name 作为可选项名称', () => { const records = flattenCategories([ { description_category_id: 17027495, category_name: '汽车用品', disabled: false, children: [ { description_category_id: 17028750, category_name: '汽车脚垫和车座套', disabled: false, children: [ { type_name: '汽车罩', type_id: 93587, disabled: false, children: [], }, ], }, ], }, ]); assert.deepEqual( records.map((record) => ({ category: record.external_category_id, type: record.external_type_id, name: record.name, path: record.path, parent: record.parent_key, active: record.is_active, })), [ { category: '17027495', type: '0', name: '汽车用品', path: '汽车用品', parent: null, active: false, }, { category: '17028750', type: '0', name: '汽车脚垫和车座套', path: '汽车用品 › 汽车脚垫和车座套', parent: '17027495:0', active: false, }, { category: '17028750', type: '93587', name: '汽车罩', path: '汽车用品 › 汽车脚垫和车座套 › 汽车罩', parent: '17028750:0', active: true, }, ], ); }); beforeAll(async () => { Object.assign(process.env, { NODE_ENV: 'test', SWAGGER_ENABLED: 'false', AUTH_REGISTER_IP_LIMIT: '100', AUTH_REGISTER_EMAIL_LIMIT: '100', }); const created = await createApplication(); app = created.app; database = app.get(DATABASE); await app.listen(0, '127.0.0.1'); baseUrl = `http://127.0.0.1:${(app.getHttpServer().address() as AddressInfo).port}`; const registration = await request('/api/auth/register', { method: 'POST', body: JSON.stringify({ email: `catalog-worker-${randomUUID()}@example.test`, password: 'correct-horse-battery-staple-123', password_confirmation: 'correct-horse-battery-staple-123', }), }); assert.equal(registration.response.status, 201); workspaceId = registration.body.data.workspace.id; userId = registration.body.data.user.id; token = registration.body.data.access_token; const platforms = await request('/api/platforms', { headers: { Authorization: `Bearer ${token}` }, }); platformId = platforms.body.data[0].id; const seeded = await database.workspaceTransaction(workspaceId, async (tx) => { const store = await tx.query<{ id: string }>( `INSERT INTO atlas.stores (workspace_id, platform_id, name, platform_client_id, currency_code, is_default, status, credential_status) VALUES (current_setting('app.workspace_id')::uuid, $1, '类目 Worker 测试店', $2, 'RUB', true, 'active', 'valid') RETURNING id`, [platformId, `catalog-worker-${randomUUID()}`], ); const secret = await tx.query<{ id: string }>( `INSERT INTO atlas.store_secrets (workspace_id, store_id, kind, secret_ciphertext, secret_nonce, encryption_key_version, fingerprint, status) VALUES (current_setting('app.workspace_id')::uuid, $1, 'seller_api_key', $2, $3, 'test', $4, 'valid') RETURNING id`, [ store.rows[0].id, Buffer.from('catalog-worker-api-key-must-not-leak'), Buffer.from('catalog-worker-nonce'), randomUUID(), ], ); return { storeId: store.rows[0].id, secretId: secret.rows[0].id }; }); storeId = seeded.storeId; secretId = seeded.secretId; }); afterAll(async () => { await database.workspaceTransaction(workspaceId, async (tx) => { await tx.query('DELETE FROM atlas.outbox_events WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.operations WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.store_product_categories WHERE workspace_id = $1', [workspaceId]); await tx.query('DELETE FROM atlas.stores WHERE workspace_id = $1', [workspaceId]); }); await database.query('DELETE FROM atlas.workspace_memberships WHERE user_id = $1', [userId]); await database.query('DELETE FROM atlas.workspaces WHERE id = $1', [workspaceId]); await database.query('DELETE FROM atlas.users WHERE id = $1', [userId]); await app.close(); }); it('空缓存的并发验证只创建一个无密钥的树同步任务', async () => { const results = await Promise.all([scheduleWithStoreLock(), scheduleWithStoreLock()]); assert.equal(results.filter((result) => result.scheduled).length, 1); assert.ok(results.some((result) => !result.scheduled && result.reason === 'already_queued')); await database.workspaceTransaction(workspaceId, async (tx) => { const events = await tx.query<{ event_payload: unknown }>( `SELECT event_payload FROM atlas.outbox_events WHERE aggregate_id = $1 AND event_type = 'store_category_tree_sync.requested'`, [storeId], ); assert.equal(events.rows.length, 1); const payload = JSON.stringify(events.rows[0].event_payload); assert.ok(!payload.includes('catalog-worker-api-key-must-not-leak')); assert.ok(!payload.includes('catalog-worker-nonce')); }); }); it('已有已同步的门店缓存时跳过 Ozon 请求调度', async () => { await database.workspaceTransaction(workspaceId, async (tx) => { await tx.query( `UPDATE atlas.operations SET status = 'succeeded', completed_at = now() WHERE kind = 'store_category_tree_sync' AND resource_id = $1`, [storeId], ); await tx.query( `INSERT INTO atlas.store_product_categories (workspace_id, store_id, platform_id, external_category_id, external_type_id, name, path, synced_at) VALUES (current_setting('app.workspace_id')::uuid, $1, $2, 'worker-cache', 'worker-type', '已缓存类目', '已缓存类目', now())`, [storeId, platformId], ); }); const result = await scheduleWithStoreLock(); assert.deepEqual(result, { scheduled: false, reason: 'cache_available' }); }); it('为 Ozon 暂时失败配置五次指数退避', async () => { const enqueued: Array<{ name: string; options: Record }> = []; const dispatcher = new OutboxDispatcher( { query: async () => ({ rows: [{ id: workspaceId }] }), workspaceTransaction: async (_workspaceId: string, work: any) => work({ query: async (sql: string) => sql.includes('RETURNING event.id') ? { rows: [ { id: randomUUID(), workspace_id: workspaceId, event_type: 'store_category_tree_sync.requested', event_payload: { store_id: storeId, secret_id: secretId, }, }, ], } : { rows: [] }, }), } as never, { add: async (name: string, _data: unknown, options: Record) => { enqueued.push({ name, options }); }, } as never, 'worker-test', ); await dispatcher.dispatchBatch(); assert.deepEqual(enqueued, [ { name: 'store_category_tree_sync.requested', options: { jobId: enqueued[0]?.options.jobId, attempts: 5, backoff: { type: 'exponential', delay: 5_000 }, removeOnComplete: 1_000, removeOnFail: 1_000, }, }, ]); }); });