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

feat(ai-image): AI生图开发

Browse Source
This commit is contained in:
joey-home
2026-09-27 22:40:52 +08:00
parent e670d06ca6
commit a29dd92676
105 changed files with 9943 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/src/main.ts
+33
View File
@@ -29,6 +29,10 @@ import {
syncPlatformCatalog,
} from './platform-catalog-sync.js';
import { runProductAiTask } from './product-ai.js';
import { markAspectRatioDerivativeFailed, runAspectRatioDerivative } from './media-aspect-ratio.js';
import { markWatermarkDerivativeFailed, runWatermarkDerivative } from './media-watermark.js';
import { classifyAiEditFailure, markMediaAiEditFailed, runMediaAiEdit } from './media-ai-edit.js';
import { archiveExpiredMissingOnlineProducts, markOnlineProductSyncFailed, runOnlineProductSync, type OnlineProductSyncJob } from './online-product-sync.js';
const connection = new Redis(workerConfig.redisUrl, {
maxRetriesPerRequest: null,
@@ -82,6 +86,18 @@ const worker = new Worker<OutboxJob | PlatformCatalogJob>(
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 === 'media.aspect_ratio_derivative.requested') {
await runAspectRatioDerivative(
database,
objectStorage,
job.data as OutboxJob,
);
} else if (job.name === 'media.watermark_derivative.requested') {
await runWatermarkDerivative(database, objectStorage, job.data as OutboxJob);
} else if (job.name === 'media.ai_edit.requested') {
await runMediaAiEdit(database, objectStorage, job.data as OutboxJob);
} else if (job.name === 'platform_product_sync.requested') {
await runOnlineProductSync(database, job.data as OnlineProductSyncJob);
} else if (job.name === 'store_credential_verification.requested') {
await verifyStoreCredentials(
database,
@@ -160,6 +176,14 @@ worker.on('failed', (job, error) => {
database,
job.data as PlatformCatalogJob,
);
} else if (job.name === 'media.aspect_ratio_derivative.requested') {
void markAspectRatioDerivativeFailed(database, job.data as OutboxJob);
} else if (job.name === 'media.watermark_derivative.requested') {
void markWatermarkDerivativeFailed(database, job.data as OutboxJob);
} else if (job.name === 'media.ai_edit.requested') {
void markMediaAiEditFailed(database, job.data as OutboxJob, classifyAiEditFailure(error));
} else if (job.name === 'platform_product_sync.requested') {
void markOnlineProductSyncFailed(database, job.data as OnlineProductSyncJob, error);
} else if (job.name === 'store_credential_verification.requested') {
void markUnavailable(database, job.data as StoreVerificationJob, error);
} else if (
@@ -376,6 +400,7 @@ async function markUnavailable(
let dispatcherTimer: ReturnType<typeof setInterval> | undefined;
let mediaCleanupTimer: ReturnType<typeof setInterval> | undefined;
let platformCatalogTimer: ReturnType<typeof setInterval> | undefined;
let onlineProductArchiveTimer: ReturnType<typeof setInterval> | undefined;
async function bootstrap() {
await queue.waitUntilReady();
await platformQueue.waitUntilReady();
@@ -384,6 +409,7 @@ async function bootstrap() {
await dispatcher.dispatchBatch();
await platformDispatcher.dispatchBatch();
await platformScheduler.run();
await archiveExpiredMissingOnlineProducts(database);
dispatcherTimer = setInterval(() => {
void dispatcher.dispatchBatch().catch((error: unknown) => {
console.error(
@@ -408,6 +434,12 @@ async function bootstrap() {
}),
60_000,
);
onlineProductArchiveTimer = setInterval(
() => void archiveExpiredMissingOnlineProducts(database).catch((error: unknown) => {
console.error('online_product_archive_failed', error instanceof Error ? error.message : error);
}),
60 * 60_000,
);
void mediaUploadCleanup.run().catch((error: unknown) => {
console.error(
'media_upload_cleanup_startup_failed',
@@ -439,6 +471,7 @@ async function shutdown() {
if (dispatcherTimer) clearInterval(dispatcherTimer);
if (mediaCleanupTimer) clearInterval(mediaCleanupTimer);
if (platformCatalogTimer) clearInterval(platformCatalogTimer);
if (onlineProductArchiveTimer) clearInterval(onlineProductArchiveTimer);
await worker.close();
await platformWorker.close();
await queue.close();
apps/worker/src/media-ai-edit.ts
+456
View File
@@ -0,0 +1,456 @@
import { createHash, randomUUID } from 'node:crypto';
import { lookup } from 'node:dns/promises';
import { request as httpsRequest } from 'node:https';
import { isIP, type LookupFunction } from 'node:net';
import { GetObjectCommand, PutObjectCommand, type S3Client } from '@aws-sdk/client-s3';
import type { MediaAiEditFailureStage, MediaAiEditSettings } from '@atlas/contracts';
import type { Database } from '@atlas/database';
import sharp from 'sharp';
import { buildMediaAiPrompt } from './media-ai-prompt.js';
import type { OutboxJob } from './outbox-dispatcher.js';
type StoredAsset = { id: string; storage_provider: string; bucket: string; object_key: string };
type EditTask = {
status: 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled';
source_asset_id: string;
reference_asset_ids: string[];
settings: MediaAiEditSettings;
provider_task_id: string | null;
created_at: Date | string;
};
const sizes: Record<MediaAiEditSettings['ratio'], [number, number]> = {
'3:4': [1152, 1536], '1:1': [1024, 1024], '4:5': [1280, 1600],
};
type AiEditFailureCode = 'AI_IMAGE_GENERATION_FAILED' | 'AI_IMAGE_CONTENT_REJECTED' | 'AI_IMAGE_PROVIDER_FAILED';
export type AiEditFailure = {
code: AiEditFailureCode;
source: 'pcore' | 'atlas';
stage: MediaAiEditFailureStage;
detail: string;
providerErrorRaw: string | null;
};
class AiEditStageError extends Error {
public constructor(public readonly stage: MediaAiEditFailureStage, public readonly cause: unknown) {
super(cause instanceof Error ? cause.message : '未知错误');
}
}
async function atStage<T>(stage: MediaAiEditFailureStage, action: () => Promise<T>): Promise<T> {
try { return await action(); }
catch (error) { throw error instanceof AiEditStageError ? error : new AiEditStageError(stage, error); }
}
/** 只保存可供排查的短摘要，避免把密钥或带签名的地址写入任务结果。 */
function safeFailureDetail(error: unknown, stage: MediaAiEditFailureStage): string {
const message = error instanceof Error ? error.message : '未知错误';
if (stage === 'task_setup' || stage === 'input_preparation' || stage === 'result_storage') {
if (['图片对象不存在', '源图片超过处理上限', '参考图片格式不受支持',
'参考图片素材不存在'].includes(message)) return message;
if (error instanceof PcoreResponseError) return message.slice(0, 500);
const code = error && typeof error === 'object' && 'code' in error && typeof error.code === 'string'
&& /^[A-Z0-9_]{2,32}$/i.test(error.code) ? `（${error.code}）` : '';
return `${error instanceof Error ? error.name : 'Error'}${code}`;
}
return message.replace(/Bearer\s+\S+/gi, 'Bearer [已隐藏]')
.replace(/https?:\/\/[^\s)]+/gi, '[地址已隐藏]')
.replace(/(?:api[_-]?key|token|secret)\s*[:=]\s*\S+/gi, '[凭据已隐藏]')
.replace(/\bai-edits\/[^\s)]+/gi, '[对象键已隐藏]')
.slice(0, 500);
}
export function classifyAiEditFailure(error: unknown): AiEditFailure {
const stage = error instanceof AiEditStageError ? error.stage : 'task_setup';
const cause = error instanceof AiEditStageError ? error.cause : error;
const providerError = cause instanceof PcoreResponseError ? cause : null;
const transportFailure = (stage === 'provider_submission' || stage === 'provider_polling')
&& cause instanceof Error && ['TypeError', 'TimeoutError', 'AbortError'].includes(cause.name);
return {
code: 'AI_IMAGE_GENERATION_FAILED',
source: providerError || transportFailure ? 'pcore' : 'atlas',
stage,
detail: safeFailureDetail(cause, stage),
providerErrorRaw: providerError?.rawResponse ?? null,
};
}
export class PcoreResponseError extends Error {
public readonly rawResponse: string | null;
public constructor(message: string, rawResponse: string | null) {
super(message);
this.rawResponse = rawResponse ? rawResponse.slice(0, 32_768) : null;
}
}
class PcoreTerminalError extends PcoreResponseError {
public constructor(public readonly errorCode: AiEditFailureCode, rawResponse: string | null) {
super(errorCode === 'AI_IMAGE_CONTENT_REJECTED'
? 'Pcore 未能完成这次生图，可能是输入或生成结果被审核拦截'
: 'Pcore 图片任务失败', rawResponse);
}
}
function pcoreErrorText(value: unknown): string {
if (typeof value === 'string') return value;
if (!value || typeof value !== 'object') return '';
const error = value as { code?: unknown; message?: unknown };
return [error.code, error.message].filter((item) => typeof item === 'string').join(' ');
}
function pcoreFailureCode(value: unknown): AiEditFailureCode {
if (!value || typeof value !== 'object') return 'AI_IMAGE_PROVIDER_FAILED';
const error = value as { code?: unknown; type?: unknown };
const explicitCode = [error.code, error.type].filter((item) => typeof item === 'string').join(' ');
return /safety|content[_-]?policy|moderation|blocked/i.test(explicitCode)
? 'AI_IMAGE_CONTENT_REJECTED' : 'AI_IMAGE_PROVIDER_FAILED';
}
function publicIpv4(address: string): boolean {
if (isIP(address) !== 4) return false;
const [a, b, c] = address.split('.').map(Number);
return !(a === 0 || a === 10 || a === 127 || a >= 224
|| a === 100 && b >= 64 && b <= 127
|| a === 169 && b === 254
|| a === 172 && b >= 16 && b <= 31
|| a === 192 && (b === 168 || b === 0 && (c === 0 || c === 2))
|| a === 198 && (b === 18 || b === 19 || b === 51 && c === 100)
|| a === 203 && b === 0 && c === 113);
}
async function readObject(storage: S3Client, asset: StoredAsset): Promise<Buffer> {
const response = await storage.send(new GetObjectCommand({ Bucket: asset.bucket, Key: asset.object_key }));
if (!response.Body) throw new Error('图片对象不存在');
const chunks: Buffer[] = [];
let bytes = 0;
for await (const chunk of response.Body as AsyncIterable<Uint8Array>) {
bytes += chunk.byteLength;
if (bytes > 20 * 1024 * 1024) throw new Error('源图片超过处理上限');
chunks.push(Buffer.from(chunk));
}
return Buffer.concat(chunks, bytes);
}
/** 普通编辑向供应商传原始图片字节，避免转码改变参考图内容。 */
async function referenceImageDataUri(bytes: Buffer): Promise<string> {
const image = sharp(bytes, { limitInputPixels: 100_000_000, failOn: 'error' });
const meta = await image.metadata();
if (!['jpeg', 'png', 'webp'].includes(meta.format ?? '') || !meta.width || !meta.height)
throw new Error('参考图片格式不受支持');
const mime = meta.format === 'jpeg' ? 'image/jpeg' : `image/${meta.format}`;
return `data:${mime};base64,${bytes.toString('base64')}`;
}
async function outpaintInputs(bytes: Buffer): Promise<{ image: Buffer; mask: Buffer; original: Buffer; left: number; top: number }> {
const [width, height] = sizes['3:4'];
const original = await sharp(bytes, { limitInputPixels: 100_000_000, failOn: 'error' })
.rotate().resize(width, height, { fit: 'inside', withoutEnlargement: false })
.ensureAlpha().png().toBuffer();
const meta = await sharp(original).metadata();
const left = Math.floor((width - meta.width!) / 2);
const top = Math.floor((height - meta.height!) / 2);
const canvas = await sharp({ create: { width, height, channels: 4, background: '#00000000' } })
.composite([{ input: original, left, top }]).png().toBuffer();
const opaque = await sharp({ create: { width: meta.width!, height: meta.height!, channels: 4, background: '#ffffffff' } })
.png().toBuffer();
const mask = await sharp({ create: { width, height, channels: 4, background: '#00000000' } })
.composite([{ input: opaque, left, top }]).png().toBuffer();
return {
image: canvas,
mask,
original, left, top,
};
}
function providerConfig(): { base: string; key: string } {
const key = process.env.PCORE_API_KEY;
const rawBase = process.env.PCORE_BASE_URL?.trim();
if (!key || !rawBase) throw new Error('PCORE_API_KEY 和 PCORE_BASE_URL 必须配置');
const url = new URL(rawBase);
const path = url.pathname.replace(/\/+$/, '');
if (url.protocol !== 'https:' || url.username || url.password || url.port
|| !['', '/v1'].includes(path) || url.search || url.hash)
throw new Error('PCORE_BASE_URL 必须是 HTTPS 根地址或 /v1 地址');
return { base: `${url.origin}/v1`, key };
}
async function providerRequest(url: string, key: string, body?: Record<string, unknown>): Promise<Record<string, unknown>> {
const response = await fetch(url, {
method: body ? 'POST' : 'GET',
headers: { Authorization: `Bearer ${key}`, ...(body ? { 'Content-Type': 'application/json' } : {}) },
body: body ? JSON.stringify(body) : undefined,
signal: AbortSignal.timeout(body ? 90_000 : 35_000),
redirect: 'error',
});
if (!response.ok) throw new PcoreResponseError(
`Pcore 图片任务请求失败（HTTP ${response.status}）`, await response.text());
let result: Record<string, unknown>;
try {
result = await response.json() as Record<string, unknown>;
if (!result || typeof result !== 'object' || Array.isArray(result)) throw new Error('invalid response');
} catch {
throw new PcoreResponseError('Pcore 返回的图片任务响应不是有效 JSON 对象', null);
}
if (result.error && (body || !['failed', 'error', 'cancelled'].includes(String(result.status ?? ''))))
throw new PcoreResponseError(`Pcore 图片任务请求被拒绝${pcoreErrorText(result.error)
? `（${pcoreErrorText(result.error).slice(0, 200)}）` : ''}`,
JSON.stringify(result.error));
return result;
}
async function submitProvider(storage: S3Client, assets: StoredAsset[], settings: MediaAiEditSettings): Promise<string> {
const provider = await atStage('provider_submission', async () => providerConfig());
const size = sizes[settings.mode === 'outpaint' ? '3:4' : settings.ratio];
const main = await atStage('input_preparation', () => readObject(storage, assets[0]));
const outpaint = settings.mode === 'outpaint'
? await atStage('input_preparation', () => outpaintInputs(main)) : null;
const images = await atStage('input_preparation', async () => outpaint
? [`data:image/png;base64,${outpaint.image.toString('base64')}`]
: await Promise.all(assets.map(async (asset, index) =>
referenceImageDataUri(index === 0 ? main : await readObject(storage, asset)))));
const body: Record<string, unknown> = {
model: 'gpt-image-2', prompt: buildMediaAiPrompt(settings), n: 1,
size: `${size[0]}x${size[1]}`, async: true,
images: images.map((image_url) => ({ image_url })),
};
if (outpaint) body.mask = { image_url: `data:image/png;base64,${outpaint.mask.toString('base64')}` };
const result = await atStage('provider_submission', () =>
providerRequest(`${provider.base}/images/edits`, provider.key, body));
const id = result.task_id ?? result.id;
if (typeof id !== 'string' || !id || id.length > 256)
throw new AiEditStageError('provider_submission', new PcoreResponseError('Pcore 未返回有效的任务 ID', null));
return id;
}
export async function downloadProviderImage(urlValue: string): Promise<Buffer> {
const url = new URL(urlValue);
if (url.protocol !== 'https:' || url.username || url.password || url.port || isIP(url.hostname))
throw new Error('Pcore 结果地址不安全');
const addresses = await lookup(url.hostname, { all: true, family: 4 });
if (!addresses.length || addresses.some((item) => !publicIpv4(item.address)))
throw new Error('Pcore 结果地址不安全');
/** 固定到校验过的公网 IPv4，同时保留域名用于 TLS 证书校验。 */
const response = await new Promise<import('node:http').IncomingMessage>((resolve, reject) => {
const request = httpsRequest(url, {
lookup: pinnedLookup(addresses[0].address),
timeout: 60_000,
}, resolve);
request.on('error', reject);
request.on('timeout', () => request.destroy(new PcoreResponseError('Pcore 结果下载超时', null)));
request.end();
});
if (response.statusCode !== 200 || Number(response.headers['content-length'] ?? 0) > 25 * 1024 * 1024) {
response.resume();
throw new PcoreResponseError(`Pcore 结果下载失败（HTTP ${response.statusCode}）`, null);
}
const chunks: Buffer[] = [];
let size = 0;
for await (const chunk of response) {
size += chunk.length;
if (size > 25 * 1024 * 1024) {
response.destroy();
throw new Error('Pcore 结果超过处理上限');
}
chunks.push(Buffer.from(chunk));
}
return Buffer.concat(chunks, size);
}
export function pinnedLookup(address: string): LookupFunction {
return (_hostname, options, callback) => {
const pinned = { address, family: 4 };
if (options.all) callback(null, [pinned]);
else callback(null, pinned.address, pinned.family);
};
}
export function providerImageResult(result: Record<string, unknown>): Buffer | string {
const first = Array.isArray(result.data) ? result.data[0] as Record<string, unknown> | undefined : undefined;
for (const encoded of [first?.b64_json, first?.b64_image]) {
if (typeof encoded !== 'string' || !encoded.trim()) continue;
if (encoded.length > 35 * 1024 * 1024) throw new Error('Pcore 结果超过处理上限');
const bytes = Buffer.from(encoded.replace(/^data:[^,]*,/, ''), 'base64');
if (bytes.length) return bytes;
}
if (typeof first?.url === 'string' && first.url) return first.url;
throw new Error('Pcore 任务完成但没有图片');
}
async function pollProvider(taskId: string): Promise<Buffer | null> {
const provider = await atStage('provider_polling', async () => providerConfig());
const result = await atStage('provider_polling', () =>
providerRequest(`${provider.base}/images/edits/${encodeURIComponent(taskId)}`, provider.key));
const status = String(result.status ?? '');
if (['failed', 'error', 'cancelled'].includes(status))
throw new PcoreTerminalError(pcoreFailureCode(result.error),
JSON.stringify(result.error ?? result));
if (!['completed', 'succeeded'].includes(status)) return null;
const image = await atStage('result_download', async () => {
try { return providerImageResult(result); }
catch (error) {
throw new PcoreResponseError(error instanceof Error ? error.message : 'Pcore 图片结果无效', null);
}
});
return typeof image === 'string'
? atStage('result_download', () => downloadProviderImage(image)) : image;
}
async function schedulePoll(database: Database, workspaceId: string, id: string) {
await database.workspaceTransaction(workspaceId, async (tx) => {
const current = await tx.query<{ status: string }>(
'SELECT status FROM atlas.media_ai_edits WHERE id = $1 FOR UPDATE', [id],
);
if (current.rows[0]?.status !== 'running') return;
await tx.query(
`INSERT INTO atlas.outbox_events
(workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key, next_attempt_at)
VALUES (current_setting('app.workspace_id')::uuid, 'media_ai_edit', $1,
'media.ai_edit.requested', jsonb_build_object('edit_id', $1::uuid), $2, now() + interval '3 seconds')`,
[id, `${id}:${randomUUID()}`],
);
});
}
export async function normalizeAiEditOutput(bytes: Buffer, settings: MediaAiEditSettings, source?: Buffer): Promise<Buffer> {
const target = sizes[settings.mode === 'outpaint' ? '3:4' : settings.ratio];
const image = sharp(bytes, { limitInputPixels: 100_000_000, failOn: 'error' });
const meta = await image.metadata();
if (!['png', 'jpeg', 'webp'].includes(meta.format ?? '') || !meta.width || !meta.height)
throw new Error('Pcore 结果图片无效');
if (settings.mode === 'outpaint' && source) {
const restored = await outpaintInputs(source);
return image.rotate().resize(target[0], target[1], { fit: 'cover' })
.composite([{ input: restored.original, left: restored.left, top: restored.top }])
.flatten({ background: '#ffffff' }).jpeg({ quality: 92 }).toBuffer();
}
return image.rotate().resize(target[0], target[1], { fit: 'contain', background: '#ffffff' })
.flatten({ background: '#ffffff' }).jpeg({ quality: 92 }).toBuffer();
}
/** 每个 Outbox 事件只推进一次供应商状态，等待由下一条延迟事件承担。 */
export async function runMediaAiEdit(database: Database, storage: S3Client, job: OutboxJob) {
const id = String(job.payload.edit_id ?? '');
const task = await atStage('task_setup', () => database.workspaceTransaction(job.workspaceId, async (tx) => {
const result = await tx.query<EditTask>(
'SELECT status, source_asset_id, reference_asset_ids, settings, provider_task_id, created_at FROM atlas.media_ai_edits WHERE id = $1 FOR UPDATE', [id],
);
const current = result.rows[0];
if (!current || !['queued', 'running'].includes(current.status)) return null;
if (Date.now() - new Date(current.created_at).getTime() > 20 * 60_000)
throw new AiEditStageError('provider_polling', new PcoreResponseError('等待 Pcore 图片任务超时', null));
await tx.query(
`UPDATE atlas.media_ai_edits SET status = 'running', started_at = coalesce(started_at, now()) WHERE id = $1`, [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 IN ('queued', 'running')`, [id],
);
return current;
}));
if (!task) return;
const assetIds = [task.source_asset_id, ...task.reference_asset_ids];
const assets = await atStage('input_preparation', () => database.workspaceTransaction(job.workspaceId, async (tx) => {
const result = await tx.query<StoredAsset>(
`SELECT id, storage_provider, bucket, object_key FROM atlas.media_assets WHERE id = ANY($1::uuid[])`, [assetIds],
);
const byId = new Map(result.rows.map((item) => [item.id, item]));
return assetIds.map((assetId) => {
const asset = byId.get(assetId);
if (!asset) throw new Error('参考图片素材不存在');
return asset;
});
}));
if (!task.provider_task_id) {
const providerTaskId = await submitProvider(storage, assets, task.settings);
await atStage('task_setup', () => database.workspaceTransaction(job.workspaceId, async (tx) => {
await tx.query(
`UPDATE atlas.media_ai_edits SET provider_task_id = $2 WHERE id = $1 AND status = 'running' AND provider_task_id IS NULL`,
[id, providerTaskId],
);
}));
await atStage('task_setup', () => schedulePoll(database, job.workspaceId, id));
return;
}
let result: Buffer | null;
try {
result = await pollProvider(task.provider_task_id);
} catch (error) {
// 供应商任务已是终态，重试同一个任务 ID 不会改变结果。
if (!(error instanceof PcoreTerminalError)) throw error;
await markMediaAiEditFailed(database, job, {
code: error.errorCode, source: 'pcore', stage: 'provider_polling',
detail: safeFailureDetail(error, 'provider_polling'), providerErrorRaw: error.rawResponse,
});
return;
}
if (!result) {
await atStage('task_setup', () => schedulePoll(database, job.workspaceId, id));
return;
}
const source = task.settings.mode === 'outpaint'
? await atStage('result_processing', () => readObject(storage, assets[0])) : undefined;
const output = await atStage('result_processing', () => normalizeAiEditOutput(result, task.settings, source));
if (output.byteLength > 20 * 1024 * 1024)
throw new AiEditStageError('result_processing', new Error('AI 编辑结果超过素材上限'));
const checksum = createHash('sha256').update(output).digest('hex');
const objectKey = `ai-edits/${job.workspaceId}/${checksum}.jpg`;
const existing = await atStage('result_storage', () => database.workspaceTransaction(job.workspaceId, async (tx) => {
const row = await tx.query<{ id: string }>('SELECT id FROM atlas.media_assets WHERE checksum_sha256 = $1', [checksum]);
return row.rows[0]?.id ?? null;
}));
if (!existing) await atStage('result_storage', () => storage.send(new PutObjectCommand({
Bucket: assets[0].bucket, Key: objectKey, Body: output, ContentType: 'image/jpeg',
})));
const [width, height] = sizes[task.settings.mode === 'outpaint' ? '3:4' : task.settings.ratio];
await atStage('result_storage', () => database.workspaceTransaction(job.workspaceId, async (tx) => {
const current = await tx.query<{ status: string }>(
'SELECT status FROM atlas.media_ai_edits WHERE id = $1 FOR UPDATE', [id],
);
if (current.rows[0]?.status !== 'running') return;
const asset = await tx.query<{ id: string }>(
`INSERT INTO atlas.media_assets
(workspace_id, storage_provider, bucket, object_key, content_type, byte_size, checksum_sha256, width_px, height_px)
VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, 'image/jpeg', $4, $5, $6, $7)
ON CONFLICT (workspace_id, checksum_sha256)
DO UPDATE SET checksum_sha256 = EXCLUDED.checksum_sha256 RETURNING id`,
[assets[0].storage_provider, assets[0].bucket, objectKey, output.byteLength, checksum, width, height],
);
const assetId = asset.rows[0].id;
await tx.query(
`UPDATE atlas.media_ai_edits SET status = 'succeeded', output_asset_id = $2, error_code = NULL,
completed_at = now() WHERE id = $1`, [id, assetId],
);
await tx.query(
`UPDATE atlas.operations SET status = 'succeeded', completed_at = now(),
result = jsonb_build_object('output_asset_id', $2::uuid), error_code = NULL, error_message = NULL
WHERE id = $1 AND status IN ('queued', 'running')`, [id, assetId],
);
}));
}
export async function markMediaAiEditFailed(
database: Database, job: OutboxJob, failure: AiEditFailure,
) {
const id = String(job.payload.edit_id ?? '');
const { code, source, stage, detail, providerErrorRaw } = failure;
const message = code === 'AI_IMAGE_CONTENT_REJECTED'
? 'Pcore 未能完成这次生图，可能是输入或生成结果被审核拦截'
: code === 'AI_IMAGE_PROVIDER_FAILED'
? 'Pcore 图片任务失败，请稍后重试'
: '单张 AI 编辑失败，请重试';
await database.workspaceTransaction(job.workspaceId, async (tx) => {
await tx.query(
`UPDATE atlas.media_ai_edits SET status = 'failed', error_code = $2,
completed_at = now() WHERE id = $1 AND status IN ('queued', 'running')`, [id, code],
);
await tx.query(
`UPDATE atlas.operations SET status = 'failed', error_code = $2,
error_message = $3,
result = coalesce(result, '{}'::jsonb) || jsonb_strip_nulls(jsonb_build_object(
'failure_source', $4::text, 'failure_stage', $5::text,
'failure_detail', $6::text, 'provider_error_raw', $7::text)),
completed_at = now()
WHERE id = $1 AND status IN ('queued', 'running')`,
[id, code, message, source, stage, detail, providerErrorRaw],
);
});
}
apps/worker/src/media-ai-prompt.ts
+47
View File
@@ -0,0 +1,47 @@
import type { MediaAiEditSettings } from '@atlas/contracts';
const styles: Record<number, string> = {
1: 'Nordic minimalist e-commerce style: warm ivory, soft daylight, natural texture and spacious composition.',
2: 'Fresh bright e-commerce style: clean high-key daylight, natural colors and an airy retail composition.',
3: 'Premium dark editorial style: charcoal setting, sculpted side lighting, realistic material texture.',
4: 'Warm lifestyle style: believable cream home setting, balanced golden and window light, product as the hero.',
5: 'Clean studio style: neutral seamless background, even softbox light, crisp edges and no decorative props.',
};
const backgrounds: Record<MediaAiEditSettings['background'], string> = {
auto: 'Keep the original background unless the user requests a change.',
white: 'Replace only the background with pure white; retain natural contact shadows.',
solid: 'Replace only the background with the solid color described by the user.',
scene: 'Replace only the background with the realistic scene described by the user.',
reference: 'Use Image 2 as background and lighting reference, but never copy its product.',
};
const languages = { ru: 'Russian', ar: 'Arabic', en: 'English' };
/** 主图是商品事实来源；辅助图仅影响背景、光线和质感。 */
export function buildMediaAiPrompt(input: MediaAiEditSettings): string {
if (input.mode === 'outpaint') return [
'Extend Image 1 to an exact 3:4 canvas. Generate only the transparent surrounding area.',
'Keep the entire original image region visually unchanged and continue the surrounding scene naturally.',
input.prompt ? `User requirements for the new edge area: ${input.prompt}` : '',
'Do not add text, watermarks, panels or new products.',
].filter(Boolean).join('\n\n');
const parts = [
input.prompt ? `User requirements (highest priority): ${input.prompt}` : '',
'Create exactly ONE finished e-commerce image. Image 1 is the fixed main product image. Preserve its shape, proportions, color, material, logo and distinguishing details. Never replace it with a product from another image.',
input.reference_asset_ids.length
? `Images 2 through ${input.reference_asset_ids.length + 1} are auxiliary references only. Use them for details, background or lighting; Image 1 remains the product source of truth.` : '',
`BACKGROUND: ${backgrounds[input.background]}`,
input.background_description ? `Background details: ${input.background_description}` : '',
input.translate_text
? `Read visible marketing text in Image 1, translate its meaning into ${languages[input.target_language]}, replace source-language marketing copy and redesign for a legible layout. For Arabic use right-to-left typesetting. Preserve physical product branding, factual values and units; add no unsupported claims.`
: 'Do not invent marketing copy or translate existing image text; preserve text on the physical product.',
];
if (input.generate_size_chart && input.dimensions) {
const dimensions = ([['length', 'length'], ['width', 'width'], ['height', 'height']] as const)
.flatMap(([key, label]) => input.dimensions?.[key]
? [`${label}: ${input.dimensions[key]} ${input.dimensions.unit}`] : []);
parts.push(`Create a clean product measurement diagram with legible arrows. Label only these supplied dimensions: ${dimensions.join(', ')}. Do not infer other measurements.`);
}
parts.push(`VISUAL STYLE: ${input.style_prompt || styles[input.style_set]}`);
parts.push(`OUTPUT: exact ${input.ratio} composition. Follow explicit user, background and text instructions before style preferences. No watermark or extra panels.`);
return parts.filter(Boolean).join('\n\n');
}
apps/worker/src/media-aspect-ratio.ts
+221
View File
@@ -0,0 +1,221 @@
import { createHash } from 'node:crypto';
import { GetObjectCommand, PutObjectCommand, type S3Client } from '@aws-sdk/client-s3';
import type { Database } from '@atlas/database';
import sharp from 'sharp';
import type { OutboxJob } from './outbox-dispatcher.js';
type Source = {
status: 'queued' | 'running' | 'succeeded' | 'failed';
storage_provider: string;
bucket: string;
object_key: string;
};
/** 读取正式素材时始终限制字节数，避免对象被外部替换后耗尽 Worker 内存。 */
async function readSource(storage: S3Client, source: Source): Promise<Buffer> {
const object = await storage.send(new GetObjectCommand({
Bucket: source.bucket,
Key: source.object_key,
}));
if (!object.Body) throw new Error('源图片不存在');
const chunks: Buffer[] = [];
let size = 0;
for await (const chunk of object.Body as AsyncIterable<Uint8Array>) {
size += chunk.byteLength;
if (size > 20 * 1024 * 1024) throw new Error('源图片超过处理上限');
chunks.push(Buffer.from(chunk));
}
return Buffer.concat(chunks, size);
}
type Edge = 'top' | 'bottom' | 'left' | 'right';
type Rgb = [number, number, number];
/** 用边缘内侧的窄带计算底色，减弱单行像素噪声对留白的影响。 */
function edgeMean(data: Buffer, width: number, height: number, edge: Edge): Rgb {
const sums: Rgb = [0, 0, 0];
let count = 0;
const depth = 8;
for (let y = 0; y < height; y++) {
for (let x = 0; x < width; x++) {
if (edge === 'top' && y >= depth) continue;
if (edge === 'bottom' && y < height - depth) continue;
if (edge === 'left' && x >= depth) continue;
if (edge === 'right' && x < width - depth) continue;
const offset = (y * width + x) * 3;
for (let channel = 0; channel < 3; channel++) sums[channel] += data[offset + channel]!;
count++;
}
}
return sums.map((sum) => Math.round(sum / count)) as Rgb;
}
/** 接缝处保持原图边缘色，短距离内平滑过渡为边缘底色。 */
function fadeWeight(distance: number): number {
const t = Math.min(Math.max(distance - 1, 0) / 5, 1);
return t * t * (3 - 2 * t);
}
/** 等比扩充到固定画布，边缘颜色平滑延展，不让纹理形成贯穿留白的条纹。 */
export async function renderThreeFour(bytes: Buffer): Promise<Buffer> {
const image = sharp(bytes, { limitInputPixels: 100_000_000, failOn: 'error' });
const metadata = await image.metadata();
if (!['jpeg', 'png', 'webp'].includes(metadata.format ?? ''))
throw new Error('源图片格式不受支持');
if (!metadata.width || !metadata.height || metadata.width > 10_000 || metadata.height > 10_000)
throw new Error('源图片尺寸不受支持');
const swapped = (metadata.orientation ?? 1) >= 5 && (metadata.orientation ?? 1) <= 8;
const width = swapped ? metadata.height : metadata.width;
const height = swapped ? metadata.width : metadata.height;
const wide = width * 4 > height * 3;
const outputWidth = wide ? 1200 : Math.max(1, Math.round(width * 1600 / height));
const outputHeight = wide ? Math.max(1, Math.round(height * 1200 / width)) : 1600;
const left = Math.floor((1200 - outputWidth) / 2);
const top = Math.floor((1600 - outputHeight) / 2);
const { data, info } = await image.rotate().resize(outputWidth, outputHeight, { fit: 'fill' })
.flatten({ background: '#ffffff' }).toColourspace('srgb').removeAlpha()
.raw().toBuffer({ resolveWithObject: true });
if (info.channels !== 3) throw new Error('图片颜色通道不受支持');
const firstMean = edgeMean(data, outputWidth, outputHeight, wide ? 'top' : 'left');
const secondMean = edgeMean(data, outputWidth, outputHeight, wide ? 'bottom' : 'right');
const canvas = Buffer.alloc(1200 * 1600 * 3);
for (let y = 0; y < 1600; y++) {
const targetStart = y * 1200 * 3;
if (wide) {
const relativeY = y - top;
if (relativeY >= 0 && relativeY < outputHeight) {
const sourceStart = relativeY * 1200 * 3;
data.copy(canvas, targetStart, sourceStart, sourceStart + 1200 * 3);
continue;
}
const firstSide = relativeY < 0;
const sourceStart = (firstSide ? 0 : outputHeight - 1) * 1200 * 3;
const mean = firstSide ? firstMean : secondMean;
const weight = fadeWeight(firstSide ? -relativeY : relativeY - outputHeight + 1);
for (let x = 0; x < 1200; x++) {
const sourceOffset = sourceStart + x * 3;
const targetOffset = targetStart + x * 3;
for (let channel = 0; channel < 3; channel++)
canvas[targetOffset + channel] = Math.round(data[sourceOffset + channel]! * (1 - weight) + mean[channel]! * weight);
}
continue;
}
const sourceStart = y * outputWidth * 3;
const firstPixel = data.subarray(sourceStart, sourceStart + 3);
const lastPixel = data.subarray(sourceStart + (outputWidth - 1) * 3, sourceStart + outputWidth * 3);
data.copy(canvas, targetStart + left * 3, sourceStart, sourceStart + outputWidth * 3);
for (let x = 0; x < left; x++) {
const weight = fadeWeight(left - x);
for (let channel = 0; channel < 3; channel++)
canvas[targetStart + x * 3 + channel] = Math.round(firstPixel[channel]! * (1 - weight) + firstMean[channel]! * weight);
}
for (let x = left + outputWidth; x < 1200; x++) {
const weight = fadeWeight(x - left - outputWidth + 1);
for (let channel = 0; channel < 3; channel++)
canvas[targetStart + x * 3 + channel] = Math.round(lastPixel[channel]! * (1 - weight) + secondMean[channel]! * weight);
}
}
return sharp(canvas, { raw: { width: 1200, height: 1600, channels: 3 } })
.jpeg({ quality: 95, chromaSubsampling: '4:4:4' }).toBuffer();
}
/** 同一派生 ID 反复执行得到同一结果；业务完成状态只保存在 PostgreSQL。 */
export async function runAspectRatioDerivative(
database: Database,
storage: S3Client,
job: OutboxJob,
) {
const derivativeId = String(job.payload.derivative_id ?? '');
const source = await database.workspaceTransaction(job.workspaceId, async (tx) => {
const result = await tx.query<Source>(
`SELECT d.status, a.storage_provider, a.bucket, a.object_key
FROM atlas.media_aspect_ratio_derivatives d
JOIN atlas.media_assets a ON a.id = d.source_asset_id
WHERE d.id = $1 FOR UPDATE OF d`,
[derivativeId],
);
const row = result.rows[0];
if (!row || row.status === 'succeeded' || row.status === 'failed') return null;
await tx.query(
`UPDATE atlas.media_aspect_ratio_derivatives
SET status = 'running', started_at = coalesce(started_at, now())
WHERE id = $1`,
[derivativeId],
);
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')`,
[derivativeId],
);
return row;
});
if (!source) return;
const output = await renderThreeFour(await readSource(storage, source));
if (output.byteLength > 20 * 1024 * 1024) throw new Error('生成图片超过处理上限');
const checksum = createHash('sha256').update(output).digest('hex');
const objectKey = `derivatives/${job.workspaceId}/three-four/${checksum}.jpg`;
const existing = await database.workspaceTransaction(job.workspaceId, async (tx) => {
const result = await tx.query<{ id: string }>(
'SELECT id FROM atlas.media_assets WHERE checksum_sha256 = $1',
[checksum],
);
return result.rows[0]?.id ?? null;
});
if (!existing) {
await storage.send(new PutObjectCommand({
Bucket: source.bucket,
Key: objectKey,
Body: output,
ContentType: 'image/jpeg',
}));
}
await database.workspaceTransaction(job.workspaceId, async (tx) => {
const asset = await tx.query<{ id: string }>(
`INSERT INTO atlas.media_assets
(workspace_id, storage_provider, bucket, object_key, content_type,
byte_size, checksum_sha256, width_px, height_px)
VALUES (current_setting('app.workspace_id')::uuid, $1,
$2, $3, 'image/jpeg', $4, $5, 1200, 1600)
ON CONFLICT (workspace_id, checksum_sha256)
DO UPDATE SET checksum_sha256 = EXCLUDED.checksum_sha256
RETURNING id`,
[source.storage_provider, source.bucket, objectKey, output.byteLength, checksum],
);
if (!asset.rows[0]) throw new Error('派生素材登记失败');
await tx.query(
`UPDATE atlas.media_aspect_ratio_derivatives
SET status = 'succeeded', output_asset_id = $2, completed_at = now(), error_code = NULL
WHERE id = $1 AND status = 'running'`,
[derivativeId, asset.rows[0].id],
);
await tx.query(
`UPDATE atlas.operations SET status = 'succeeded', completed_at = now(),
result = jsonb_build_object('output_asset_id', $2::uuid),
error_code = NULL, error_message = NULL
WHERE id = $1 AND status IN ('queued', 'running')`,
[derivativeId, asset.rows[0].id],
);
});
}
/** 重试耗尽后只向前端暴露稳定错误码，内部异常不会进入任务记录。 */
export async function markAspectRatioDerivativeFailed(database: Database, job: OutboxJob) {
const derivativeId = String(job.payload.derivative_id ?? '');
await database.workspaceTransaction(job.workspaceId, async (tx) => {
await tx.query(
`UPDATE atlas.media_aspect_ratio_derivatives
SET status = 'failed', error_code = 'IMAGE_PROCESSING_FAILED', completed_at = now()
WHERE id = $1 AND status IN ('queued', 'running')`,
[derivativeId],
);
await tx.query(
`UPDATE atlas.operations
SET status = 'failed', error_code = 'IMAGE_PROCESSING_FAILED',
error_message = '图片处理失败，请重试', completed_at = now()
WHERE id = $1 AND status IN ('queued', 'running')`,
[derivativeId],
);
});
}
apps/worker/src/media-watermark.ts
+218
View File
@@ -0,0 +1,218 @@
import { createHash } from 'node:crypto';
import { GetObjectCommand, PutObjectCommand, type S3Client } from '@aws-sdk/client-s3';
import type { Database } from '@atlas/database';
import sharp from 'sharp';
import type { OutboxJob } from './outbox-dispatcher.js';
type StoredObject = { storage_provider: string; bucket: string; object_key: string };
type Task = StoredObject & {
status: 'queued' | 'running' | 'succeeded' | 'failed';
kind: 'text' | 'image';
text_content: string | null;
font_code: string | null;
font_size_px: number | null;
opacity: string;
scale_ratio: string | null;
image_size_px: number | null;
border_radius_px: number | null;
position_x: number | null;
position_y: number | null;
position: string;
watermark_bucket: string | null;
watermark_object_key: string | null;
};
/** 对象存储内容在读入内存时再次限制大小，不依赖上传时的校验。 */
async function readObject(storage: S3Client, bucket: string, key: string, maxBytes: number): Promise<Buffer> {
const object = await storage.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
if (!object.Body) throw new Error('图片对象不存在');
const chunks: Buffer[] = [];
let size = 0;
for await (const chunk of object.Body as AsyncIterable<Uint8Array>) {
size += chunk.byteLength;
if (size > maxBytes) throw new Error('图片对象超过处理上限');
chunks.push(Buffer.from(chunk));
}
return Buffer.concat(chunks, size);
}
function escapeXml(value: string): string {
return value.replace(/[&<>"']/g, (character) => ({
'&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&apos;',
})[character]!);
}
function legacyCoordinate(position: string): [number, number] {
const values: Record<string, [number, number]> = {
top_left: [0, 0], top_center: [0.5, 0], top_right: [1, 0],
middle_left: [0, 0.5], center: [0.5, 0.5], middle_right: [1, 0.5],
bottom_left: [0, 1], bottom_center: [0.5, 1], bottom_right: [1, 1],
};
return values[position] ?? [1, 1];
}
function placement(canvas: number, mark: number, coordinate: number, requestedInset: number): number {
const inset = Math.min(requestedInset, Math.max(0, Math.floor((canvas - mark) / 2)));
return Math.max(0, Math.min(canvas - mark, Math.round(inset + coordinate * Math.max(0, canvas - mark - 2 * inset))));
}
/** 渲染器只使用已锁定的模板版本；透明度作用于整个水印图层。 */
export async function renderWatermark(sourceBytes: Buffer, template: Task, watermarkBytes?: Buffer): Promise<{
bytes: Buffer; width: number; height: number;
}> {
const input = sharp(sourceBytes, { limitInputPixels: 100_000_000, failOn: 'error' });
const meta = await input.metadata();
if (!['jpeg', 'png', 'webp'].includes(meta.format ?? '') || !meta.width || !meta.height
|| meta.width > 10_000 || meta.height > 10_000) throw new Error('源图片格式或尺寸不受支持');
const rotated = (meta.orientation ?? 1) >= 5 && (meta.orientation ?? 1) <= 8;
const width = rotated ? meta.height : meta.width;
const height = rotated ? meta.width : meta.height;
/** 模板中的像素值以 240×320 预览框为基准，等比映射到完整源图。 */
const previewScale = Math.max(1, width / 240, height / 320);
const inset = Math.max(1, Math.round(5 * previewScale));
let layer: Buffer;
if (template.kind === 'image') {
if (!watermarkBytes) throw new Error('图片水印素材不存在');
const overlay = sharp(watermarkBytes, { limitInputPixels: 25_000_000, failOn: 'error' });
const overlayMeta = await overlay.metadata();
if (!['jpeg', 'png', 'webp'].includes(overlayMeta.format ?? '') || !overlayMeta.width || !overlayMeta.height)
throw new Error('图片水印格式不受支持');
const baseSize = Math.min(200, Math.max(10, template.image_size_px ?? Math.round(Number(template.scale_ratio ?? 0.5) * 144)));
const overlayRotated = (overlayMeta.orientation ?? 1) >= 5 && (overlayMeta.orientation ?? 1) <= 8;
const overlayWidth = overlayRotated ? overlayMeta.height : overlayMeta.width;
const overlayHeight = overlayRotated ? overlayMeta.width : overlayMeta.height;
const ratio = overlayWidth / overlayHeight;
const desiredWidth = baseSize * previewScale * Math.min(1, ratio);
const desiredHeight = baseSize * previewScale / Math.max(1, ratio);
const fit = Math.min(1, Math.max(1, width - 2 * inset) / desiredWidth,
Math.max(1, height - 2 * inset) / desiredHeight);
const markWidth = Math.max(1, Math.round(desiredWidth * fit));
const markHeight = Math.max(1, Math.round(desiredHeight * fit));
const resized = await overlay.rotate().resize(markWidth, markHeight, { fit: 'fill' }).ensureAlpha().png().toBuffer();
const radius = Math.max(0, Math.round((template.border_radius_px ?? 0) * previewScale * fit));
if (radius) {
const resizedMeta = await sharp(resized).metadata();
const mask = Buffer.from(`<svg width="${resizedMeta.width}" height="${resizedMeta.height}"><rect width="100%" height="100%" rx="${radius}" ry="${radius}" fill="white"/></svg>`);
layer = await sharp(resized).composite([{ input: mask, blend: 'dest-in' }]).png().toBuffer();
} else layer = resized;
} else {
const content = template.text_content ?? '';
const baseFontSize = template.font_size_px ?? 16;
if (!content || content.length > 200 || baseFontSize < 12 || baseFontSize > 60) throw new Error('文字水印配置无效');
const fontSize = Math.min(1000, Math.max(1, Math.round(baseFontSize * previewScale)));
const textContent = content.replace(/\s+/gu, ' ');
const safeFont = escapeXml(template.font_code ?? 'sans-serif');
const svgWidth = Math.min(6000, Math.max(100, Math.ceil(textContent.length * fontSize * 1.2 + 20)));
const svgHeight = Math.max(100, Math.ceil(fontSize * 1.6 + 20));
const svg = `<svg xmlns="http://www.w3.org/2000/svg" width="${svgWidth}" height="${svgHeight}"><text x="5" y="${fontSize + 5}" font-family="${safeFont},sans-serif" font-size="${fontSize}" font-weight="600" fill="#111827">${escapeXml(textContent)}</text></svg>`;
layer = await sharp(Buffer.from(svg)).trim().png().toBuffer();
}
const opacity = Number(template.opacity);
if (!Number.isFinite(opacity) || opacity < 0 || opacity > 1) throw new Error('水印透明度无效');
const { data, info } = await sharp(layer).ensureAlpha().raw().toBuffer({ resolveWithObject: true });
if (info.channels !== 4) throw new Error('水印通道无效');
for (let index = 3; index < data.length; index += 4) data[index] = Math.round(data[index]! * opacity);
const maxWidth = Math.max(1, width - 2 * inset);
const maxHeight = Math.max(1, height - 2 * inset);
layer = await sharp(data, { raw: { width: info.width, height: info.height, channels: 4 } })
.resize({ width: maxWidth, height: maxHeight, fit: 'inside', withoutEnlargement: true })
.png().toBuffer();
const layerMeta = await sharp(layer).metadata();
const legacy = legacyCoordinate(template.position);
const x = template.position_x ?? legacy[0];
const y = template.position_y ?? legacy[1];
if (x < 0 || x > 1 || y < 0 || y > 1) throw new Error('水印坐标无效');
const composite = {
input: layer,
left: placement(width, layerMeta.width!, x, inset),
top: placement(height, layerMeta.height!, y, inset),
};
/** 大幅面源图可能生成超过素材上限的 JPEG，逐级压缩但不改变画布尺寸。 */
for (const quality of [92, 82, 72, 62, 50]) {
const bytes = await input.clone().rotate().flatten({ background: '#ffffff' })
.composite([composite]).jpeg({ quality, chromaSubsampling: '4:4:4' }).toBuffer();
if (bytes.byteLength <= 20 * 1024 * 1024) return { bytes, width, height };
}
throw new Error('水印图片超过处理上限');
}
/** 同一任务可安全重试，正式素材以工作空间和内容哈希去重。 */
export async function runWatermarkDerivative(database: Database, storage: S3Client, job: OutboxJob) {
const id = String(job.payload.derivative_id ?? '');
const task = await database.workspaceTransaction(job.workspaceId, async (tx) => {
const result = await tx.query<Task>(
`SELECT d.status, source.storage_provider, source.bucket, source.object_key,
v.kind, v.text_content, v.font_code, v.font_size_px, v.opacity::text,
v.scale_ratio::text, v.image_size_px, v.border_radius_px,
v.position_x::float8, v.position_y::float8, v.position::text,
mark.bucket AS watermark_bucket, mark.object_key AS watermark_object_key
FROM atlas.media_watermark_derivatives d
JOIN atlas.media_assets source ON source.id = d.source_asset_id
JOIN atlas.watermark_template_versions v
ON v.workspace_id = d.workspace_id AND v.template_id = d.watermark_template_id
AND v.version = d.watermark_template_version
LEFT JOIN atlas.media_assets mark ON mark.id = v.image_asset_id
WHERE d.id = $1 FOR UPDATE OF d`,
[id],
);
const row = result.rows[0];
if (!row || row.status === 'succeeded' || row.status === 'failed') return null;
await tx.query(
`UPDATE atlas.media_watermark_derivatives
SET status = 'running', started_at = coalesce(started_at, now()) WHERE id = $1`, [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 IN ('queued', 'running')`, [id]);
return row;
});
if (!task) return;
const sourceBytes = await readObject(storage, task.bucket, task.object_key, 20 * 1024 * 1024);
const markBytes = task.kind === 'image' && task.watermark_bucket && task.watermark_object_key
? await readObject(storage, task.watermark_bucket, task.watermark_object_key, 5 * 1024 * 1024)
: undefined;
const output = await renderWatermark(sourceBytes, task, markBytes);
if (output.bytes.byteLength > 20 * 1024 * 1024) throw new Error('水印图片超过处理上限');
const checksum = createHash('sha256').update(output.bytes).digest('hex');
const objectKey = `derivatives/${job.workspaceId}/watermark/${checksum}.jpg`;
const existing = await database.workspaceTransaction(job.workspaceId, async (tx) => {
const result = await tx.query<{ id: string }>('SELECT id FROM atlas.media_assets WHERE checksum_sha256 = $1', [checksum]);
return result.rows[0]?.id ?? null;
});
if (!existing) await storage.send(new PutObjectCommand({
Bucket: task.bucket, Key: objectKey, Body: output.bytes, ContentType: 'image/jpeg',
}));
await database.workspaceTransaction(job.workspaceId, async (tx) => {
const asset = await tx.query<{ id: string }>(
`INSERT INTO atlas.media_assets
(workspace_id, storage_provider, bucket, object_key, content_type,
byte_size, checksum_sha256, width_px, height_px)
VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3, 'image/jpeg', $4, $5, $6, $7)
ON CONFLICT (workspace_id, checksum_sha256)
DO UPDATE SET checksum_sha256 = EXCLUDED.checksum_sha256 RETURNING id`,
[task.storage_provider, task.bucket, objectKey, output.bytes.byteLength, checksum, output.width, output.height],
);
const assetId = asset.rows[0]?.id;
if (!assetId) throw new Error('水印素材登记失败');
await tx.query(
`UPDATE atlas.media_watermark_derivatives SET status = 'succeeded', output_asset_id = $2,
completed_at = now(), error_code = NULL WHERE id = $1 AND status = 'running'`, [id, assetId]);
await tx.query(
`UPDATE atlas.operations SET status = 'succeeded', completed_at = now(),
result = jsonb_build_object('output_asset_id', $2::uuid), error_code = NULL, error_message = NULL
WHERE id = $1 AND status IN ('queued', 'running')`, [id, assetId]);
});
}
/** 重试耗尽后只暴露稳定错误码。 */
export async function markWatermarkDerivativeFailed(database: Database, job: OutboxJob) {
const id = String(job.payload.derivative_id ?? '');
await database.workspaceTransaction(job.workspaceId, async (tx) => {
await tx.query(
`UPDATE atlas.media_watermark_derivatives SET status = 'failed', error_code = 'IMAGE_PROCESSING_FAILED',
completed_at = now() WHERE id = $1 AND status IN ('queued', 'running')`, [id]);
await tx.query(
`UPDATE atlas.operations SET status = 'failed', error_code = 'IMAGE_PROCESSING_FAILED',
error_message = '水印处理失败，请重试', completed_at = now()
WHERE id = $1 AND status IN ('queued', 'running')`, [id]);
});
}
apps/worker/src/online-product-mapper.ts
+155
View File
@@ -0,0 +1,155 @@
export type OzonProductListItem = {
product_id: number | string;
offer_id: string;
};
export type OzonProductDetail = Record<string, unknown>;
export type MappedOnlineProduct = {
remote_product_id: string;
offer_id: string;
remote_sku: string | null;
name: string | null;
selling_price: string | null;
list_price: string | null;
visible: boolean | null;
remote_state: string | null;
moderation_status: string | null;
validation_status: string | null;
stock_summary: { total: number; has_data: boolean };
image_summary: {
url: string | null;
width_px: number | null;
height_px: number | null;
};
error_payload: Array<{ code?: string; message: string }>;
visibility_details: Record<string, unknown>;
raw_payload: OzonProductDetail;
};
export function parseOzonListPage(value: unknown): {
items: OzonProductListItem[];
lastId: string;
} {
const result = record(record(value)?.result);
if (!result || !Array.isArray(result.items))
throw new Error('Ozon 商品列表响应格式异常');
const items = result.items.map((raw) => {
const item = record(raw);
const productId = identifier(item?.product_id);
const offerId = text(item?.offer_id);
if (
!productId ||
!/^\d+$/.test(productId) ||
!Number.isSafeInteger(Number(productId)) ||
!offerId
)
throw new Error('Ozon 商品列表缺少商品标识');
return { product_id: productId, offer_id: offerId };
});
return { items, lastId: text(result.last_id) ?? '' };
}
export function parseOzonInfoItems(value: unknown): OzonProductDetail[] {
const result = record(record(value)?.result);
if (!result || !Array.isArray(result.items))
throw new Error('Ozon 商品详情响应格式异常');
return result.items.filter((item): item is OzonProductDetail =>
Boolean(record(item)),
);
}
/** 只把列表需要的状态和脱敏错误摘要放入可读字段；完整平台响应仅留在受控快照。 */
export function mapOzonProduct(
detail: OzonProductDetail,
listed: OzonProductListItem,
): MappedOnlineProduct {
const status = record(detail.status);
const stock = record(detail.stocks);
const present = finiteNumber(stock?.present);
const sku = Array.isArray(detail.sources)
? (detail.sources
.map((source) => identifier(record(source)?.sku))
.find(Boolean) ?? null)
: null;
const images = Array.isArray(detail.images) ? detail.images : [];
const firstImage =
url(detail.primary_image) ??
images
.map((item) => url(item) ?? url(record(item)?.file_name))
.find(Boolean) ??
null;
const rawErrors = Array.isArray(detail.errors) ? detail.errors : [];
const declineReasons = Array.isArray(status?.decline_reasons)
? status.decline_reasons
: [];
const errors = [...rawErrors, ...declineReasons]
.slice(0, 10)
.map((item) => {
const source = record(item);
return {
...(text(source?.code) ? { code: text(source?.code)! } : {}),
message: (
text(source?.message) ??
text(source?.name) ??
text(item) ??
'平台未提供具体原因'
).slice(0, 300),
};
});
return {
remote_product_id:
identifier(detail.id) ?? identifier(listed.product_id)!,
offer_id: text(detail.offer_id) ?? listed.offer_id,
remote_sku: sku,
name: text(detail.name),
selling_price: amount(detail.price),
list_price: amount(detail.old_price),
visible: typeof detail.visible === 'boolean' ? detail.visible : null,
remote_state: text(status?.state) ?? text(detail.state),
moderation_status: text(status?.moderate_status),
validation_status: text(status?.validation_state),
stock_summary: { total: present ?? 0, has_data: present !== null },
image_summary: { url: firstImage, width_px: null, height_px: null },
error_payload: errors,
visibility_details: record(detail.visibility_details) ?? {},
raw_payload: detail,
};
}
function record(value: unknown): Record<string, unknown> | null {
return value && typeof value === 'object' && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
function text(value: unknown): string | null {
if (typeof value !== 'string') return null;
return value.trim() || null;
}
function identifier(value: unknown): string | null {
if (typeof value === 'number' && Number.isSafeInteger(value) && value >= 0)
return String(value);
return text(value);
}
function finiteNumber(value: unknown): number | null {
const parsed =
typeof value === 'number'
? value
: typeof value === 'string' && value.trim()
? Number(value)
: NaN;
return Number.isFinite(parsed) && parsed >= 0 ? parsed : null;
}
function amount(value: unknown): string | null {
const raw = typeof value === 'number' ? String(value) : text(value);
return raw && /^\d{1,14}(?:\.\d{1,4})?$/.test(raw) ? raw : null;
}
function url(value: unknown): string | null {
const candidate = text(value);
if (!candidate) return null;
try {
const parsed = new URL(candidate);
return parsed.protocol === 'https:' ? parsed.toString() : null;
} catch {
return null;
}
}
apps/worker/src/online-product-sync.ts
+346
View File
@@ -0,0 +1,346 @@
import type { Database, DatabaseTransaction } from '@atlas/database';
import type { OutboxJob } from './outbox-dispatcher.js';
import { decryptStoreSecret } from './store-secret.crypto.js';
import { workerConfig } from './config.js';
import {
OzonCredentialError,
OzonRequestError,
requestOzon,
} from './ozon-client.js';
import {
mapOzonProduct,
parseOzonInfoItems,
parseOzonListPage,
type MappedOnlineProduct,
type OzonProductListItem,
} from './online-product-mapper.js';
export type OnlineProductSyncJob = OutboxJob & {
payload: {
sync_run_id: string;
store_id: string;
platform_id: string;
secret_id: string;
};
};
type Source = {
platform_code: string;
platform_client_id: string;
secret_ciphertext: Buffer;
secret_nonce: Buffer;
};
export async function runOnlineProductSync(
database: Database,
job: OnlineProductSyncJob,
): Promise<void> {
const source = await database.workspaceTransaction(
job.workspaceId,
async (tx) => {
const run = await tx.query<{ status: string }>(
`SELECT status FROM atlas.platform_product_sync_runs WHERE id = $1 AND store_id = $2 AND platform_id = $3 FOR UPDATE`,
[
job.payload.sync_run_id,
job.payload.store_id,
job.payload.platform_id,
],
);
if (
!run.rows[0] ||
run.rows[0].status === 'succeeded' ||
run.rows[0].status === 'failed' ||
run.rows[0].status === 'cancelled'
)
return null;
const connection = await tx.query<Source>(
`SELECT platform.code AS platform_code, store.platform_client_id,
secret.secret_ciphertext, secret.secret_nonce
FROM atlas.stores store
JOIN atlas.platforms platform ON platform.id = store.platform_id
JOIN atlas.store_secrets secret ON secret.id = $3 AND secret.store_id = store.id
WHERE store.id = $1 AND store.platform_id = $2 AND store.deleted_at IS NULL
AND store.credential_status = 'valid' AND secret.revoked_at IS NULL
AND secret.kind = 'seller_api_key'`,
[
job.payload.store_id,
job.payload.platform_id,
job.payload.secret_id,
],
);
if (!connection.rows[0]) throw new Error('店铺凭据已变更或不可用');
// 同一任务重试时清除上次尝试的已见快照，避免把本次未返回的商品误判为仍在售。
await tx.query(
`DELETE FROM atlas.platform_listing_sync_snapshots WHERE sync_run_id = $1`,
[job.payload.sync_run_id],
);
await tx.query(
`UPDATE atlas.platform_product_sync_runs SET status = 'running', started_at = COALESCE(started_at, now()),
completed_at = NULL, error_payload = NULL, listed_count = 0, upserted_count = 0,
failed_count = 0, cursor_started = NULL, cursor_finished = NULL WHERE id = $1`,
[job.payload.sync_run_id],
);
return connection.rows[0];
},
);
if (!source) return;
if (source.platform_code.toUpperCase() !== 'OZON')
throw new Error('当前平台不支持商品同步');
const apiKey = decryptStoreSecret(
source.secret_ciphertext,
source.secret_nonce,
workerConfig.storeSecretEncryptionKey,
);
let cursor = '';
let listedCount = 0;
let upsertedCount = 0;
const seenCursors = new Set<string>();
for (;;) {
const page = parseOzonListPage(
await requestOzon(
source.platform_client_id,
apiKey,
'/v3/product/list',
{
filter: { visibility: 'ALL' },
last_id: cursor,
limit: 1000,
},
),
);
listedCount += page.items.length;
if (listedCount > 1_000_000)
throw new Error('商品数量超过单次同步安全上限');
for (let start = 0; start < page.items.length; start += 100) {
const batch = page.items.slice(start, start + 100);
const body = await requestOzon(
source.platform_client_id,
apiKey,
'/v3/product/info/list',
{
product_id: batch.map((item) => Number(item.product_id)),
},
);
const details = parseOzonInfoItems(body);
const detailById = new Map(
details.map((detail) => [String(detail.id ?? ''), detail]),
);
const mapped = batch.map((item) => {
const detail = detailById.get(String(item.product_id));
if (!detail) throw new Error('Ozon 商品详情缺少列表中的商品');
return mapOzonProduct(detail, item);
});
await database.workspaceTransaction(job.workspaceId, async (tx) => {
await upsertBatch(tx, job, mapped);
upsertedCount += mapped.length;
await tx.query(
`UPDATE atlas.platform_product_sync_runs SET listed_count = $2, upserted_count = $3, cursor_finished = $4
WHERE id = $1 AND status = 'running'`,
[
job.payload.sync_run_id,
listedCount,
upsertedCount,
page.lastId,
],
);
});
}
if (!page.lastId) break;
if (page.lastId === cursor || seenCursors.has(page.lastId))
throw new Error('Ozon 商品列表游标重复');
seenCursors.add(page.lastId);
cursor = page.lastId;
await database.workspaceTransaction(job.workspaceId, (tx) =>
tx.query(
`UPDATE atlas.platform_product_sync_runs SET cursor_finished = $2, listed_count = $3 WHERE id = $1 AND status = 'running'`,
[job.payload.sync_run_id, cursor, listedCount],
),
);
}
await database.workspaceTransaction(job.workspaceId, async (tx) => {
const run = await tx.query<{ status: string }>(
`SELECT status FROM atlas.platform_product_sync_runs WHERE id = $1 FOR UPDATE`,
[job.payload.sync_run_id],
);
if (run.rows[0]?.status !== 'running') return;
// 只有完整成功的全量同步可以改变未见与归档状态；失败批次保留这些标记。
await tx.query(
`UPDATE atlas.platform_listings listing SET missing_since = NULL, archived_at = NULL, updated_at = now()
WHERE listing.store_id = $1 AND listing.platform_id = $2
AND (listing.missing_since IS NOT NULL OR listing.archived_at IS NOT NULL)
AND EXISTS (
SELECT 1 FROM atlas.platform_listing_sync_snapshots snapshot
WHERE snapshot.sync_run_id = $3 AND snapshot.platform_listing_id = listing.id
)`,
[job.payload.store_id, job.payload.platform_id, job.payload.sync_run_id],
);
await tx.query(
`UPDATE atlas.platform_listings listing SET
archived_at = CASE WHEN missing_since IS NOT NULL THEN COALESCE(archived_at, now()) ELSE archived_at END,
missing_since = COALESCE(missing_since, now()), updated_at = now()
WHERE listing.store_id = $1 AND listing.platform_id = $2 AND listing.archived_at IS NULL
AND NOT EXISTS (
SELECT 1 FROM atlas.platform_listing_sync_snapshots snapshot
WHERE snapshot.sync_run_id = $3 AND snapshot.platform_listing_id = listing.id
)`,
[
job.payload.store_id,
job.payload.platform_id,
job.payload.sync_run_id,
],
);
await tx.query(
`UPDATE atlas.platform_product_sync_runs SET status = 'succeeded', listed_count = $2,
upserted_count = $3, cursor_finished = $4, completed_at = now(), error_payload = NULL
WHERE id = $1 AND status = 'running'`,
[job.payload.sync_run_id, listedCount, upsertedCount, cursor],
);
});
}
async function upsertBatch(
tx: DatabaseTransaction,
job: OnlineProductSyncJob,
items: MappedOnlineProduct[],
) {
if (!items.length) return;
if (new Set(items.map((item) => item.offer_id)).size !== items.length)
throw new Error('商品批次存在重复的商家编码');
// 平台商品 ID 变化时，沿用同店铺的商家编码所对应的镜像记录和快照历史。
await tx.query(
`UPDATE atlas.platform_listings listing SET remote_product_id = incoming.remote_product_id
FROM jsonb_to_recordset($3::jsonb) AS incoming(remote_product_id text, offer_id text)
WHERE listing.store_id = $1 AND listing.platform_id = $2
AND listing.offer_id = incoming.offer_id
AND listing.remote_product_id IS DISTINCT FROM incoming.remote_product_id`,
[
job.payload.store_id,
job.payload.platform_id,
JSON.stringify(items.map(({ remote_product_id, offer_id }) => ({ remote_product_id, offer_id }))),
],
);
const result = await tx.query<{ id: string; remote_product_id: string }>(
`WITH incoming AS (
SELECT * FROM jsonb_to_recordset($4::jsonb) AS p(
remote_product_id text, offer_id text, remote_sku text, name text,
selling_price text, list_price text, visible boolean, remote_state text,
moderation_status text, validation_status text, stock_summary jsonb,
image_summary jsonb, error_payload jsonb, visibility_details jsonb, raw_payload jsonb)
), prepared AS (
SELECT incoming.*, origin.id AS atlas_variant_listing_id
FROM incoming LEFT JOIN LATERAL (
SELECT id FROM atlas.product_variant_platform_listings
WHERE store_id = $2 AND platform_id = $3 AND status = 'imported'
AND (offer_id = incoming.offer_id OR platform_product_id = incoming.remote_product_id)
ORDER BY (offer_id = incoming.offer_id) DESC LIMIT 1
) origin ON true
)
INSERT INTO atlas.platform_listings (
workspace_id, store_id, platform_id, remote_product_id, offer_id, remote_sku,
name, currency_code, selling_price, list_price, visible, remote_state,
moderation_status, validation_status, stock_summary, image_summary, error_payload,
visibility_details, source, atlas_variant_listing_id, last_sync_run_id,
last_synced_at, raw_payload)
SELECT current_setting('app.workspace_id')::uuid, $2, $3, remote_product_id, offer_id,
remote_sku, name, 'RUB', selling_price::numeric, list_price::numeric, visible,
remote_state, moderation_status, validation_status, stock_summary, image_summary,
error_payload, visibility_details,
CASE WHEN atlas_variant_listing_id IS NULL THEN 'external_sync'::atlas.platform_listing_source
ELSE 'atlas_published'::atlas.platform_listing_source END,
atlas_variant_listing_id, $1, now(), raw_payload
FROM prepared WHERE true
ON CONFLICT (workspace_id, store_id, platform_id, remote_product_id) DO UPDATE SET
offer_id = EXCLUDED.offer_id, remote_sku = EXCLUDED.remote_sku, name = EXCLUDED.name,
currency_code = EXCLUDED.currency_code, selling_price = EXCLUDED.selling_price,
list_price = EXCLUDED.list_price, visible = EXCLUDED.visible, remote_state = EXCLUDED.remote_state,
moderation_status = EXCLUDED.moderation_status, validation_status = EXCLUDED.validation_status,
stock_summary = EXCLUDED.stock_summary, image_summary = EXCLUDED.image_summary,
error_payload = EXCLUDED.error_payload, visibility_details = EXCLUDED.visibility_details,
source = EXCLUDED.source, atlas_variant_listing_id = EXCLUDED.atlas_variant_listing_id,
last_sync_run_id = EXCLUDED.last_sync_run_id, last_synced_at = EXCLUDED.last_synced_at,
raw_payload = EXCLUDED.raw_payload, updated_at = now()
RETURNING id, remote_product_id`,
[
job.payload.sync_run_id,
job.payload.store_id,
job.payload.platform_id,
JSON.stringify(items),
],
);
const idByRemoteId = new Map(
result.rows.map((row) => [row.remote_product_id, row.id]),
);
await tx.query(
`INSERT INTO atlas.platform_listing_sync_snapshots (workspace_id, platform_listing_id, sync_run_id, payload)
SELECT current_setting('app.workspace_id')::uuid, listing.id, $1, snapshot.payload
FROM jsonb_to_recordset($2::jsonb) AS snapshot(remote_product_id text, payload jsonb)
JOIN atlas.platform_listings listing ON listing.store_id = $3 AND listing.remote_product_id = snapshot.remote_product_id
ON CONFLICT (workspace_id, platform_listing_id, sync_run_id) DO UPDATE SET payload = EXCLUDED.payload`,
[
job.payload.sync_run_id,
JSON.stringify(
items.map((item) => ({
remote_product_id: item.remote_product_id,
payload: item.raw_payload,
})),
),
job.payload.store_id,
],
);
if (idByRemoteId.size !== items.length)
throw new Error('商品批次存在重复的平台 ID');
}
export async function markOnlineProductSyncFailed(
database: Database,
job: OnlineProductSyncJob,
error: unknown,
) {
const code =
error instanceof OzonCredentialError
? 'STORE_CREDENTIAL_UNAVAILABLE'
: error instanceof OzonRequestError
? 'OZON_REQUEST_REJECTED'
: 'OZON_SYNC_UNAVAILABLE';
const message =
code === 'STORE_CREDENTIAL_UNAVAILABLE'
? '店铺凭据不可用于读取 Ozon 商品'
: code === 'OZON_REQUEST_REJECTED'
? 'Ozon 拒绝了商品读取请求，请检查店铺权限'
: '商品同步失败，请稍后重试';
await database.workspaceTransaction(job.workspaceId, (tx) =>
tx.query(
`UPDATE atlas.platform_product_sync_runs SET status = 'failed', failed_count = GREATEST(failed_count, 1),
completed_at = now(), error_payload = $2 WHERE id = $1 AND status IN ('queued', 'running')`,
[job.payload.sync_run_id, { code, message }],
),
);
}
/** 首次未见超过七天即归档；活跃全量同步期间等待其结果，避免临时隐藏正在恢复的商品。 */
export async function archiveExpiredMissingOnlineProducts(
database: Database,
workspaceId?: string,
): Promise<void> {
const workspaces = workspaceId
? [{ id: workspaceId }]
: (
await database.query<{ id: string }>(
'SELECT id FROM atlas.workspaces',
)
).rows;
for (const workspace of workspaces) {
await database.workspaceTransaction(workspace.id, (tx) =>
tx.query(
`UPDATE atlas.platform_listings listing
SET archived_at = now(), updated_at = now()
WHERE listing.missing_since <= now() - interval '7 days'
AND listing.archived_at IS NULL
AND NOT EXISTS (
SELECT 1 FROM atlas.platform_product_sync_runs run
WHERE run.store_id = listing.store_id AND run.status IN ('queued', 'running')
)`,
),
);
}
}
apps/worker/src/ozon-client.ts
+1 -1
View File
@@ -168,7 +168,7 @@ export async function fetchOzonDictionaryBatch(
return { values, isComplete: false };
}
async function requestOzon(
export async function requestOzon(
clientId: string,
apiKey: string,
path: string,