feat(storage): storage port and minio adapter with forbid-overwrite semantics

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-17 20:25:18 +08:00
parent 59464bd73d
commit 37f4b3e11f
10 changed files with 748 additions and 3 deletions
@@ -0,0 +1,142 @@
+// packages/storage/src/minio.adapter.ts —— StoragePort 的 MinIO 实现（S3 兼容协议，
+// AWS SDK v3 + path-style）。生产 OSS 适配器（P5/G1）实现同端口，行为契约以此处注释为准。
+import {
+  CopyObjectCommand,
+  GetObjectCommand,
+  HeadObjectCommand,
+  PutObjectCommand,
+  S3Client,
+} from '@aws-sdk/client-s3';
+import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
+import { Readable } from 'node:stream';
+import { ErrorCodes, StorageError } from './errors';
+import type { StoragePort, UploadCredential } from './port';
+
+export interface MinioStorageConfig {
+  endpoint: string;
+  region: string;
+  accessKey: string;
+  secretKey: string;
+  bucket: string;
+}
+
+/** S3 错误统一判定：HeadObject 对缺失键返回 404/NotFound（无 body 错误码），GetObject 为 NoSuchKey。 */
+function isNotFound(error: unknown): boolean {
+  if (typeof error !== 'object' || error === null) return false;
+  const e = error as { name?: string; $metadata?: { httpStatusCode?: number } };
+  return e.$metadata?.httpStatusCode === 404 || e.name === 'NotFound' || e.name === 'NoSuchKey';
+}
+
+export class MinioStorageAdapter implements StoragePort {
+  private readonly client: S3Client;
+  private readonly bucket: string;
+
+  constructor(config: MinioStorageConfig) {
+    const missing = (Object.keys(config) as Array<keyof MinioStorageConfig>).filter((k) => !config[k]);
+    if (missing.length > 0) {
+      throw new Error(`MinioStorageAdapter: 存储配置缺失 ${missing.join(', ')}（S3_* 环境变量见仓库根 .env）`);
+    }
+    this.bucket = config.bucket;
+    this.client = new S3Client({
+      endpoint: config.endpoint,
+      region: config.region,
+      credentials: { accessKeyId: config.accessKey, secretAccessKey: config.secretKey },
+      forcePathStyle: true,
+    });
+  }
+
+  /**
+   * 预签名上传。S3 预签名无法对 Content-Length 签名（SigV4 规范排除该头），
+   * bytes 参数仅作为端口契约保留：字节数核验发生在上传完成后的 head/verify 阶段。
+   * Content-Type 经 PutObjectCommand 的 ContentType 参数参与签名，客户端必须
+   * 原样携带返回的 headers，否则 403 SignatureDoesNotMatch。
+   */
+  async presignPut(objectKey: string, bytes: number, contentType: string, ttlSeconds: number): Promise<UploadCredential> {
+    void bytes; // 见上：签名不含 Content-Length，大小核验由 verify 阶段负责
+    const url = await getSignedUrl(
+      this.client,
+      new PutObjectCommand({ Bucket: this.bucket, Key: objectKey, ContentType: contentType }),
+      { expiresIn: ttlSeconds },
+    );
+    return {
+      mode: 'presigned',
+      url,
+      method: 'PUT',
+      headers: { 'Content-Type': contentType },
+      expiresAt: new Date(Date.now() + ttlSeconds * 1000).toISOString(),
+    };
+  }
+
+  async presignGet(objectKey: string, ttlSeconds: number): Promise<string> {
+    return getSignedUrl(this.client, new GetObjectCommand({ Bucket: this.bucket, Key: objectKey }), {
+      expiresIn: ttlSeconds,
+    });
+  }
+
+  async head(objectKey: string): Promise<{ bytes: number; etag: string } | null> {
+    try {
+      const out = await this.client.send(new HeadObjectCommand({ Bucket: this.bucket, Key: objectKey }));
+      return { bytes: out.ContentLength ?? 0, etag: out.ETag ?? '' };
+    } catch (error) {
+      if (isNotFound(error)) return null;
+      throw error;
+    }
+  }
+
+  async getStream(objectKey: string): Promise<NodeJS.ReadableStream> {
+    const out = await this.client.send(new GetObjectCommand({ Bucket: this.bucket, Key: objectKey }));
+    if (!out.Body) {
+      throw new StorageError(ErrorCodes.UPLOAD_VERIFY_FAILED, `getStream: 对象无 Body: ${objectKey}`);
+    }
+    return out.Body as Readable;
+  }
+
+  async put(objectKey: string, body: Buffer, contentType: string): Promise<{ etag: string }> {
+    const out = await this.client.send(
+      new PutObjectCommand({ Bucket: this.bucket, Key: objectKey, ContentType: contentType, Body: body }),
+    );
+    return { etag: out.ETag ?? '' };
+  }
+
+  /**
+   * 禁止覆盖语义的复制（staging → 正式键）：
+   * - dest 已存在且字节数与 expectedBytes 一致 → 幂等返回 created:false（同 etag）；
+   * - dest 已存在但字节数不符 → StorageError(UPLOAD_VERIFY_FAILED)；
+   * - dest 不存在 → CopyObject（MetadataDirective=COPY，MinIO 需显式指定以保留源元数据）
+   *   后 head 复核字节数再返回 created:true。
+   *
+   * 已知竞态窗口：MinIO 的 CopyObject 不支持条件写（无 forbid-overwrite 头），两个并发
+   * copyIfAbsent 同时通过 head 检查后，后发的 CopyObject 会覆盖先发的结果。本项目接受该
+   * 窗口：同一会话的并发调用复制的是同一个 staging 源对象，覆盖结果字节一致（幂等等价）；
+   * 跨会话撞键理论上不可能（正式键含会话/照片唯一标识）。生产 OSS 适配器将使用
+   * x-oss-forbid-overwrite 头消除该窗口（G1 验收点）。
+   */
+  async copyIfAbsent(src: string, dest: string, expectedBytes: number): Promise<{ created: boolean; etag: string }> {
+    const existing = await this.head(dest);
+    if (existing) {
+      if (existing.bytes === expectedBytes) return { created: false, etag: existing.etag };
+      throw new StorageError(
+        ErrorCodes.UPLOAD_VERIFY_FAILED,
+        `copyIfAbsent: dest 已存在且字节数不符（expected=${expectedBytes}, actual=${existing.bytes}）: ${dest}`,
+      );
+    }
+    // CopySource 须为 "bucket/key" 且 key 部分 URL 编码（保留 / 分隔符）
+    const copySource = [this.bucket, ...src.split('/').map(encodeURIComponent)].join('/');
+    await this.client.send(
+      new CopyObjectCommand({
+        Bucket: this.bucket,
+        Key: dest,
+        CopySource: copySource,
+        MetadataDirective: 'COPY',
+      }),
+    );
+    const verified = await this.head(dest);
+    if (!verified || verified.bytes !== expectedBytes) {
+      throw new StorageError(
+        ErrorCodes.UPLOAD_VERIFY_FAILED,
+        `copyIfAbsent: 复制后复核失败（对象缺失或字节数不符）: ${dest}`,
+      );
+    }
+    return { created: true, etag: verified.etag };
+  }
+}