atlas-seller-kit/apps/api/test/worker-catalog-scheduler.e2e.test.ts

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 { 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 调度集成测试', () => {
    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>(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<string, unknown> }> = [];
        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<string, unknown>) => {
                    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,
                },
            },
        ]);
    });
});