joey5628/atlas-seller-kit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai-edit): 开发 AI 填报功能

Browse Source
This commit is contained in:
R524809
2026-09-23 17:18:09 +08:00
parent 95a2a37f1b
commit b3958166dd
35 changed files with 2832 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/src/main.ts
+105 -37
View File
@@ -19,23 +19,34 @@ import {
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 {
PlatformCatalogDispatcher,
type PlatformCatalogJob,
} from './platform-catalog-dispatcher.js';
import { PlatformCatalogScheduler } from './platform-catalog-scheduler.js';
import { markPlatformCatalogUnavailable, syncPlatformCatalog } from './platform-catalog-sync.js';
import {
markPlatformCatalogUnavailable,
syncPlatformCatalog,
} from './platform-catalog-sync.js';
import { runProductAiTask } from './product-ai.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 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()}`,
database,
platformQueue,
`platform_worker_${randomUUID()}`,
);
const platformScheduler = new PlatformCatalogScheduler(database);
const objectStorage = new S3Client({
@@ -48,7 +59,9 @@ const objectStorage = new S3Client({
},
});
const mediaUploadCleanup = new MediaUploadCleanup(database, objectStorage);
type StoreVerificationJob = OutboxJob & { payload: { store_id: string; operation_id: string; secret_id?: string } };
type StoreVerificationJob = OutboxJob & {
payload: { store_id: string; operation_id: string; secret_id?: string };
};
type CatalogSyncJob = OutboxJob & {
payload: {
store_id: string;
@@ -67,12 +80,39 @@ const worker = new Worker<OutboxJob | PlatformCatalogJob>(
async (job) => {
if (job.name === 'platform_catalog_sync.requested') {
await syncPlatformCatalog(database, job.data as PlatformCatalogJob);
} else if (job.name === 'product_ai_task.requested') {
await runProductAiTask(database, job.data as OutboxJob);
} 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 } });
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);
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' ||
@@ -91,18 +131,35 @@ const platformWorker = new Worker<PlatformCatalogJob>(
{ connection, concurrency: 5 },
);
async function cleanupCollection(database: Database, job: OutboxJob & { payload: { operation_id: string; collection_item_ids: string[] } }) {
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]);
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);
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 (
@@ -110,7 +167,11 @@ worker.on('failed', (job, error) => {
job.name === 'store_category_attribute_sync.requested' ||
job.name === 'store_attribute_dictionary_sync.requested'
) {
void markCatalogSyncUnavailable(database, job.data as CatalogSyncJob, error);
void markCatalogSyncUnavailable(
database,
job.data as CatalogSyncJob,
error,
);
}
});
platformWorker.on('failed', (job) => {
@@ -118,7 +179,10 @@ platformWorker.on('failed', (job) => {
void markPlatformCatalogUnavailable(database, job.data);
});
async function verifyStoreCredentials(database: Database, job: StoreVerificationJob) {
async function verifyStoreCredentials(
database: Database,
job: StoreVerificationJob,
) {
const source = await database.workspaceTransaction(
job.workspaceId,
async (tx) => {
@@ -167,7 +231,10 @@ async function verifyStoreCredentials(database: Database, job: StoreVerification
}
}
async function cancelStaleOperation(database: Database, job: StoreVerificationJob) {
async function cancelStaleOperation(
database: Database,
job: StoreVerificationJob,
) {
await database.workspaceTransaction(job.workspaceId, (tx) =>
tx.query(
`UPDATE atlas.operations SET status = 'cancelled', completed_at = now(),
@@ -317,27 +384,28 @@ async function bootstrap() {
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,
);
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);
}),
() =>
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) => {