feat(api-watermark): 水印接口优化

2026-09-04 15:58:53 +08:00
parent 009082d0f4
commit f635359216
8 changed files with 7350 additions and 5436 deletions
@@ -13,6 +13,7 @@
    "dependencies": {
        "@atlas/contracts": "workspace:*",
        "@atlas/database": "workspace:*",
+        "@aws-sdk/client-s3": "3.1015.0",
        "bullmq": "6.3.4",
        "ioredis": "6.0.0"
    },
@@ -2,7 +2,11 @@ import { existsSync } from 'node:fs';
 import { resolve } from 'node:path';
 import { loadEnvFile } from 'node:process';

-for (const directory of [process.cwd(), resolve(process.cwd(), '..'), resolve(process.cwd(), '../..')]) {
+for (const directory of [
+    process.cwd(),
+    resolve(process.cwd(), '..'),
+    resolve(process.cwd(), '../..'),
+]) {
    const path = resolve(directory, '.env');
    if (existsSync(path)) {
        loadEnvFile(path);
@@ -16,8 +20,27 @@ function required(name: string): string {
    return value;
 }

+function objectStorageCredential(
+    accessKeyName: string,
+    localFallbackName: string,
+): string {
+    return process.env[accessKeyName] || required(localFallbackName);
+}
+
 export const workerConfig = {
    databaseUrl: required('DATABASE_URL'),
    redisUrl: required('REDIS_URL'),
    storeSecretEncryptionKey: required('STORE_SECRET_ENCRYPTION_KEY'),
+    objectStorageEndpoint:
+        process.env.OBJECT_STORAGE_ENDPOINT ?? 'http://localhost:9000',
+    objectStorageForcePathStyle:
+        (process.env.OBJECT_STORAGE_FORCE_PATH_STYLE ?? 'true') === 'true',
+    objectStorageAccessKey: objectStorageCredential(
+        'OBJECT_STORAGE_ACCESS_KEY',
+        'MINIO_ROOT_USER',
+    ),
+    objectStorageSecretKey: objectStorageCredential(
+        'OBJECT_STORAGE_SECRET_KEY',
+        'MINIO_ROOT_PASSWORD',
+    ),
 };
@@ -1,16 +1,38 @@
 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 {
+    OzonCredentialError,
+    OzonRequestError,
+    verifyOzonCredentials,
+} from './ozon-client.js';
 import { decryptStoreSecret } from './store-secret.crypto.js';
+import { MediaUploadCleanup } from './media-upload-cleanup.js';

-const connection = new Redis(workerConfig.redisUrl, { maxRetriesPerRequest: null });
+const connection = new Redis(workerConfig.redisUrl, {
+    maxRetriesPerRequest: null,
+});
 const database = createApplicationDatabase(workerConfig.databaseUrl);
 const queue = new Queue<OutboxJob>('atlas-dispatch', { connection });
-const dispatcher = new OutboxDispatcher(database, queue, `worker_${randomUUID()}`);
+const dispatcher = new OutboxDispatcher(
+    database,
+    queue,
+    `worker_${randomUUID()}`,
+);
+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);

 const worker = new Worker<OutboxJob>(
    'atlas-dispatch',
@@ -28,14 +50,19 @@ worker.on('failed', (job, error) => {
 });

 async function verifyStoreCredentials(database: Database, job: OutboxJob) {
-    const source = await database.workspaceTransaction(job.workspaceId, async (tx) => {
+    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;
+                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
@@ -45,11 +72,17 @@ async function verifyStoreCredentials(database: Database, job: OutboxJob) {
                [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);
+    const apiKey = decryptStoreSecret(
+        source.secret_ciphertext,
+        source.secret_nonce,
+        workerConfig.storeSecretEncryptionKey,
+    );
    try {
-        if (source.platform_code.toUpperCase() !== 'OZON') throw new OzonRequestError(422);
+        if (source.platform_code.toUpperCase() !== 'OZON')
+            throw new OzonRequestError(422);
        await verifyOzonCredentials(source.platform_client_id, apiKey);
        await completeVerification(database, job, true);
    } catch (error) {
@@ -66,12 +99,14 @@ async function verifyStoreCredentials(database: Database, job: OutboxJob) {
 }

 async function cancelStaleOperation(database: Database, job: OutboxJob) {
-    await database.workspaceTransaction(job.workspaceId, (tx) => tx.query(
+    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(
@@ -103,9 +138,13 @@ async function completeVerification(
                   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] : [
+            valid
+                ? [job.payload.store_id]
+                : [
                      job.payload.store_id,
-                rejectedStatus === 403 ? 'OZON_CREDENTIAL_FORBIDDEN' : 'OZON_CREDENTIAL_UNAUTHORIZED',
+                      rejectedStatus === 403
+                          ? 'OZON_CREDENTIAL_FORBIDDEN'
+                          : 'OZON_CREDENTIAL_UNAUTHORIZED',
                      rejectedStatus === 403
                          ? 'Ozon 拒绝访问：请检查 Client ID 与 API Key 是否属于同一账户，及密钥是否具备读取权限'
                          : 'Ozon 未接受该 Client ID 或 API Key',
@@ -119,9 +158,13 @@ async function completeVerification(
                : `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] : [
+            valid
+                ? [job.payload.operation_id]
+                : [
                      job.payload.operation_id,
-                rejectedStatus === 403 ? 'OZON_CREDENTIAL_FORBIDDEN' : 'OZON_CREDENTIAL_UNAUTHORIZED',
+                      rejectedStatus === 403
+                          ? 'OZON_CREDENTIAL_FORBIDDEN'
+                          : 'OZON_CREDENTIAL_UNAUTHORIZED',
                      rejectedStatus === 403
                          ? 'Ozon 拒绝访问：请检查 Client ID 与 API Key 是否属于同一账户，及密钥是否具备读取权限'
                          : 'Ozon 未接受该 Client ID 或 API Key',
@@ -130,7 +173,11 @@ async function completeVerification(
    });
 }

-async function markRequestRejected(database: Database, job: OutboxJob, status: number) {
+async function markRequestRejected(
+    database: Database,
+    job: OutboxJob,
+    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
@@ -156,7 +203,11 @@ async function markRequestRejected(database: Database, job: OutboxJob, status: n
    });
 }

-async function markUnavailable(database: Database, job: OutboxJob, _error: Error) {
+async function markUnavailable(
+    database: Database,
+    job: OutboxJob,
+    _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
@@ -180,23 +231,57 @@ async function markUnavailable(database: Database, job: OutboxJob, _error: Error
    });
 }

-let timer: ReturnType<typeof setInterval> | undefined;
+let dispatcherTimer: ReturnType<typeof setInterval> | undefined;
+let mediaCleanupTimer: ReturnType<typeof setInterval> | undefined;
 async function bootstrap() {
    await queue.waitUntilReady();
    await worker.waitUntilReady();
    await dispatcher.dispatchBatch();
-    timer = setInterval(() => void dispatcher.dispatchBatch().catch((error: unknown) => {
-        console.error('outbox_dispatch_failed', error instanceof Error ? error.message : error);
-    }), 2_000);
-    console.info('Atlas Worker is ready; consuming transactional-outbox events.');
+    dispatcherTimer = setInterval(
+        () =>
+            void dispatcher.dispatchBatch().catch((error: unknown) => {
+                console.error(
+                    'outbox_dispatch_failed',
+                    error instanceof Error ? error.message : error,
+                );
+            }),
+        2_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);
+    console.error(
+        'worker_startup_failed',
+        error instanceof Error ? error.message : error,
+    );
    process.exitCode = 1;
 });
 async function shutdown() {
-    if (timer) clearInterval(timer);
-    await worker.close(); await queue.close(); await database.close(); await connection.quit();
+    if (dispatcherTimer) clearInterval(dispatcherTimer);
+    if (mediaCleanupTimer) clearInterval(mediaCleanupTimer);
+    await worker.close();
+    await queue.close();
+    objectStorage.destroy();
+    await database.close();
+    await connection.quit();
 }
 process.once('SIGINT', () => void shutdown());
 process.once('SIGTERM', () => void shutdown());
@@ -0,0 +1,107 @@
+import { DeleteObjectCommand, type S3Client } from '@aws-sdk/client-s3';
+import type { Database } from '@atlas/database';
+
+type CleanupCandidate = {
+    id: string;
+    bucket: string;
+    object_key: string;
+};
+
+export class MediaUploadCleanup {
+    private running: Promise<void> | null = null;
+
+    public constructor(
+        private readonly database: Database,
+        private readonly s3: S3Client,
+    ) {}
+
+    /**
+     * 上传 URL 到期并不代表浏览器一定上传过对象；DeleteObject 对不存在对象也安全，
+     * 所以终态意图统一尝试删除。只在删除成功后写 object_cleaned_at，失败下轮重试。
+     */
+    public async run() {
+        if (this.running) return this.running;
+        this.running = this.runOnce().finally(() => {
+            this.running = null;
+        });
+        return this.running;
+    }
+
+    private async runOnce() {
+        const workspaces = await this.database.query<{ id: string }>(
+            'SELECT id FROM atlas.workspaces',
+        );
+        let expired = 0;
+        let cleaned = 0;
+        for (const workspace of workspaces.rows) {
+            expired += await this.expirePending(workspace.id);
+            const candidates = await this.claimCandidates(workspace.id);
+            for (const candidate of candidates) {
+                try {
+                    await this.s3.send(
+                        new DeleteObjectCommand({
+                            Bucket: candidate.bucket,
+                            Key: candidate.object_key,
+                        }),
+                    );
+                    await this.markCleaned(workspace.id, candidate.id);
+                    cleaned += 1;
+                } catch (error) {
+                    console.warn(
+                        'media_upload_cleanup_failed',
+                        JSON.stringify({
+                            workspace_id: workspace.id,
+                            upload_id: candidate.id,
+                            reason:
+                                error instanceof Error
+                                    ? error.message
+                                    : 'unknown_error',
+                        }),
+                    );
+                }
+            }
+        }
+        if (expired || cleaned) {
+            console.info(
+                'media_upload_cleanup_completed',
+                JSON.stringify({ expired, cleaned }),
+            );
+        }
+    }
+
+    private async expirePending(workspaceId: string): Promise<number> {
+        return this.database.workspaceTransaction(workspaceId, async (tx) => {
+            const result = await tx.query(
+                `UPDATE atlas.media_upload_intents
+                 SET status = 'expired', error_code = 'UPLOAD_EXPIRED',
+                     error_message = '上传意图已过期', completed_at = now()
+                 WHERE status = 'pending' AND expires_at <= now()`,
+            );
+            return result.rowCount ?? 0;
+        });
+    }
+
+    private async claimCandidates(workspaceId: string) {
+        return this.database.workspaceTransaction(workspaceId, async (tx) => {
+            const candidates = await tx.query<CleanupCandidate>(
+                `SELECT id, bucket, object_key
+                 FROM atlas.media_upload_intents
+                 WHERE status IN ('failed', 'expired') AND object_cleaned_at IS NULL
+                 ORDER BY created_at
+                 LIMIT 50`,
+            );
+            return candidates.rows;
+        });
+    }
+
+    private async markCleaned(workspaceId: string, intentId: string) {
+        await this.database.workspaceTransaction(workspaceId, (tx) =>
+            tx.query(
+                `UPDATE atlas.media_upload_intents SET object_cleaned_at = now()
+                 WHERE id = $1 AND status IN ('failed', 'expired')
+                   AND object_cleaned_at IS NULL`,
+                [intentId],
+            ),
+        );
+    }
+}
@@ -67,7 +67,7 @@ Headers：`Idempotency-Key` 必填。
 }
 ```

-`object_key` 是服务端内部字段，不放在普通客户端响应。上传意图和完成状态均持久化在数据库，不能只存在于 Redis 或预签名 token 中；过期和失败对象由后续清理任务回收。
+`object_key` 是服务端内部字段，不放在普通客户端响应。上传意图和完成状态均持久化在数据库，不能只存在于 Redis 或预签名 token 中；Worker 每 5 分钟将逾期未完成意图标记为 `expired`，并删除 `failed` / `expired` 意图的临时对象。删除成功才记录清理时间；失败会在下一轮重试。

 ### 1.3 临时预览 URL

@@ -72,6 +72,12 @@
 - 已实现素材短时预览 URL（10 分钟）；普通 API 返回永不包含 bucket、object key 或长期 URL。图片模板只接受当前工作空间内、类型为 PNG/JPEG/WebP 的已验证素材。
 - 本地环境默认使用 `MINIO_ROOT_USER` / `MINIO_ROOT_PASSWORD` 连接 `http://localhost:9000`；生产必须显式配置最小权限的 `OBJECT_STORAGE_ACCESS_KEY` / `OBJECT_STORAGE_SECRET_KEY`。

+### 1.7 开发进度（2026-09-04）：上传意图清理 Worker
+
+- Worker 启动时立即执行、之后每 5 分钟扫描一次所有工作空间：将超过有效期仍为 `pending` 的上传意图置为 `expired`。
+- 对 `failed` / `expired` 意图，Worker 删除其 MinIO/S3 临时对象；对象不存在也视为已清理。仅对象存储删除成功才写入 `object_cleaned_at`，删除暂时失败会保留记录并在下次扫描重试。
+- 新增迁移 `007_media_upload_cleanup.sql`，该字段与清理索引必须在部署 Worker 前执行。
+
 ## 2. 已确认产品决策

 | 主题 | 决策 |
@@ -135,7 +141,7 @@ User → Workspace → Membership / Role
 1. 验证并补齐 Outbox Dispatcher 与凭据验证 Worker 的集成测试：覆盖成功、401/403、限流/超时重试、旧密钥任务取消和无密钥日志。
 2. 在凭据验证通过后实现仓库同步 Worker、幂等 upsert 和状态回写，并补充失败重试、死信与集成测试。
 3. ~~Web 接入真实认证和门店管理 API~~ 已基本完成：登录/注册、会话刷新、平台上下文、门店列表/创建/编辑/默认切换/删除均已上线（见 1.2）；剩余水印绑定入口待素材/水印模块交付后接入。前端不得保存 API Key 或 refresh token。
-4. 接着实现素材直传和水印管理，再开始采集箱与商品编辑；浏览器扩展仅在采集 API 稳定后实现实际解析器。
+4. ~~接着实现素材直传和水印管理~~ 已完成后端上传、校验、预览与临时对象清理；接着实现水印实际渲染、采集箱与商品编辑。浏览器扩展仅在采集 API 稳定后实现实际解析器。
 5. 在开始商品编辑/在线同步模块前，关闭评审文档中的 P1 事项：同商品归属约束、未映射属性持久化、乐观锁 Repository 规则、平台能力 Schema；并确认在线商品归档策略和 Client ID 编辑边界。

 账户资料的已确认补充：`users.avatar_asset_id` 只保存素材引用，绝不保存公开 URL；`users.revision` 用于 `PATCH /api/auth/me` 的 `If-Match`。资料更新已实现幂等与版本冲突检测；修改密码已实现旧密码校验、按 IP/用户限流、撤销旧会话并轮换新会话。头像上传必须复用素材模块的预签名直传与校验流程，不引入 base64 直传；素材模块尚未实现时头像只能为 `null`。
@@ -0,0 +1,9 @@
+-- 上传意图到达终态后，Worker 需要可靠记录临时对象是否已从对象存储清理。
+SET search_path TO atlas, public;
+
+ALTER TABLE media_upload_intents
+    ADD COLUMN object_cleaned_at TIMESTAMPTZ;
+
+CREATE INDEX ix_media_upload_intents_cleanup
+    ON media_upload_intents (workspace_id, status, created_at)
+    WHERE status IN ('failed', 'expired') AND object_cleaned_at IS NULL;