atlas-seller-kit/apps/worker/src/main.ts

import { randomUUID } from 'node:crypto';
import { S3Client } from '@aws-sdk/client-s3';
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';
import { MediaUploadCleanup } from './media-upload-cleanup.js';
import { materializeCollectionAssets } from './collection-assets.js';
import {
    markCatalogSyncUnavailable,
    syncOzonCatalog,
    type CatalogSyncScope,
} from './ozon-catalog-sync.js';
import { scheduleStoreCategoryTreeSync } from './store-category-tree-scheduler.js';
import { PlatformCatalogDispatcher, type PlatformCatalogJob } from './platform-catalog-dispatcher.js';
import { PlatformCatalogScheduler } from './platform-catalog-scheduler.js';
import { markPlatformCatalogUnavailable, syncPlatformCatalog } from './platform-catalog-sync.js';

const connection = new Redis(workerConfig.redisUrl, {
    maxRetriesPerRequest: null,
});
const database = createApplicationDatabase(workerConfig.databaseUrl);
const queue = new Queue<OutboxJob>('atlas-dispatch', { connection });
const platformQueue = new Queue<PlatformCatalogJob>('atlas-platform-catalog', { connection });
const dispatcher = new OutboxDispatcher(
    database,
    queue,
    `worker_${randomUUID()}`,
);
const platformDispatcher = new PlatformCatalogDispatcher(
    database, platformQueue, `platform_worker_${randomUUID()}`,
);
const platformScheduler = new PlatformCatalogScheduler(database);
const objectStorage = new S3Client({
    endpoint: workerConfig.objectStorageEndpoint,
    region: 'us-east-1',
    forcePathStyle: workerConfig.objectStorageForcePathStyle,
    credentials: {
        accessKeyId: workerConfig.objectStorageAccessKey,
        secretAccessKey: workerConfig.objectStorageSecretKey,
    },
});
const mediaUploadCleanup = new MediaUploadCleanup(database, objectStorage);
type StoreVerificationJob = OutboxJob & { payload: { store_id: string; operation_id: string; secret_id?: string } };
type CatalogSyncJob = OutboxJob & {
    payload: {
        store_id: string;
        secret_id: string;
        operation_id: string;
        platform_id: string;
        scope: CatalogSyncScope;
        category_id?: string;
        dictionary_id?: string;
        query?: string;
    };
};

const worker = new Worker<OutboxJob | PlatformCatalogJob>(
    'atlas-dispatch',
    async (job) => {
        if (job.name === 'platform_catalog_sync.requested') {
            await syncPlatformCatalog(database, job.data as PlatformCatalogJob);
        } else if (job.name === 'store_credential_verification.requested') {
            await verifyStoreCredentials(database, job.data as OutboxJob & { payload: { store_id: string; operation_id: string; secret_id?: string } });
        } else if (job.name === 'collection.cleanup.requested') {
            await cleanupCollection(database, job.data as OutboxJob & { payload: { operation_id: string; collection_item_ids: string[] } });
        } else if (job.name === 'collection.assets.materialize.requested' || job.name === 'collection.assets.retry.requested') {
            await materializeCollectionAssets(database, objectStorage, workerConfig.objectStorageBucket, job.data as OutboxJob);
        } else if (
            job.name === 'store_category_tree_sync.requested' ||
            job.name === 'store_category_attribute_sync.requested' ||
            job.name === 'store_attribute_dictionary_sync.requested'
        ) {
            await syncOzonCatalog(database, job.data as CatalogSyncJob);
        }
    },
    { connection, concurrency: 5 },
);

// 平台目录使用独立队列，交互型高优先级刷新不会与素材任务共用等待队列。
const platformWorker = new Worker<PlatformCatalogJob>(
    'atlas-platform-catalog',
    async (job) => syncPlatformCatalog(database, job.data),
    { connection, concurrency: 5 },
);

async function cleanupCollection(database: Database, job: OutboxJob & { payload: { operation_id: string; collection_item_ids: string[] } }) {
    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 = 'queued'`, [job.payload.operation_id]);
        await tx.query(`UPDATE atlas.collection_asset_transfers SET status = 'cancelled', updated_at = now(), completed_at = now() WHERE collection_item_id = ANY($1::uuid[]) AND status IN ('queued', 'running')`, [job.payload.collection_item_ids]);
        await tx.query(`UPDATE atlas.operations SET status = 'succeeded', completed_at = now(), result = jsonb_build_object('cancelled_transfer_count', 0) WHERE id = $1 AND status IN ('queued', 'running')`, [job.payload.operation_id]);
    });
}

worker.on('failed', (job, error) => {
    if (!job || job.attemptsMade < (job.opts.attempts ?? 1)) return;
    if (job.name === 'platform_catalog_sync.requested') {
        void markPlatformCatalogUnavailable(database, job.data as PlatformCatalogJob);
    } else if (job.name === 'store_credential_verification.requested') {
        void markUnavailable(database, job.data as StoreVerificationJob, error);
    } else if (
        job.name === 'store_category_tree_sync.requested' ||
        job.name === 'store_category_attribute_sync.requested' ||
        job.name === 'store_attribute_dictionary_sync.requested'
    ) {
        void markCatalogSyncUnavailable(database, job.data as CatalogSyncJob, error);
    }
});
platformWorker.on('failed', (job) => {
    if (!job || job.attemptsMade < (job.opts.attempts ?? 1)) return;
    void markPlatformCatalogUnavailable(database, job.data);
});

async function verifyStoreCredentials(database: Database, job: StoreVerificationJob) {
    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: StoreVerificationJob) {
    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: StoreVerificationJob,
    valid: boolean,
    rejectedStatus?: 401 | 403,
) {
    await database.workspaceTransaction(job.workspaceId, async (tx) => {
        const current = await tx.query<{ id: string; platform_id: string }>(
            `SELECT s.id, s.platform_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',
                  ],
        );
        if (!valid) return;
        await scheduleStoreCategoryTreeSync(tx, {
            storeId: job.payload.store_id,
            platformId: current.rows[0].platform_id,
            secretId: job.payload.secret_id ?? '',
        });
    });
}

async function markRequestRejected(
    database: Database,
    job: StoreVerificationJob,
    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: StoreVerificationJob,
    _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 dispatcherTimer: ReturnType<typeof setInterval> | undefined;
let mediaCleanupTimer: ReturnType<typeof setInterval> | undefined;
let platformCatalogTimer: ReturnType<typeof setInterval> | undefined;
async function bootstrap() {
    await queue.waitUntilReady();
    await platformQueue.waitUntilReady();
    await worker.waitUntilReady();
    await platformWorker.waitUntilReady();
    await dispatcher.dispatchBatch();
    await platformDispatcher.dispatchBatch();
    await platformScheduler.run();
    dispatcherTimer = setInterval(
        () => {
            void dispatcher.dispatchBatch().catch((error: unknown) => {
                console.error(
                    'outbox_dispatch_failed',
                    error instanceof Error ? error.message : error,
                );
            });
            void platformDispatcher.dispatchBatch().catch((error: unknown) => {
                console.error(
                    'platform_outbox_dispatch_failed',
                    error instanceof Error ? error.message : error,
                );
            });
        },
        2_000,
    );
    platformCatalogTimer = setInterval(
        () => void platformScheduler.run().catch((error: unknown) => {
            console.error('platform_catalog_schedule_failed', error instanceof Error ? error.message : error);
        }),
        60_000,
    );
    void mediaUploadCleanup.run().catch((error: unknown) => {
        console.error(
            'media_upload_cleanup_startup_failed',
            error instanceof Error ? error.message : error,
        );
    });
    mediaCleanupTimer = setInterval(
        () =>
            void mediaUploadCleanup.run().catch((error: unknown) => {
                console.error(
                    'media_upload_cleanup_failed',
                    error instanceof Error ? error.message : error,
                );
            }),
        5 * 60_000,
    );
    console.info(
        'Atlas Worker is ready; consuming transactional-outbox events.',
    );
}
void bootstrap().catch((error: unknown) => {
    console.error(
        'worker_startup_failed',
        error instanceof Error ? error.message : error,
    );
    process.exitCode = 1;
});
async function shutdown() {
    if (dispatcherTimer) clearInterval(dispatcherTimer);
    if (mediaCleanupTimer) clearInterval(mediaCleanupTimer);
    if (platformCatalogTimer) clearInterval(platformCatalogTimer);
    await worker.close();
    await platformWorker.close();
    await queue.close();
    await platformQueue.close();
    objectStorage.destroy();
    await database.close();
    await connection.quit();
}
process.once('SIGINT', () => void shutdown());
process.once('SIGTERM', () => void shutdown());