photo-live/packages/storage/src/minio.adapter.ts

// 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;
  /** 寻址风格：true=path-style（MinIO 默认）；false=virtual-host（OSS/AWS 云形态）。 */
  forcePathStyle?: boolean;
}

/** 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) {
    // 只校验必填字符串（forcePathStyle 是可选布尔，false≠缺失）。
    const required = ['endpoint', 'region', 'accessKey', 'secretKey', 'bucket'] as const;
    const missing = required.filter((k) => !config[k]);
    if (missing.length > 0) {
      throw new Error(`MinioStorageAdapter: 存储配置缺失 ${missing.join(', ')}（S3_* 环境变量见仓库根 .env）`);
    }
    this.bucket = config.bucket;
    // 寻址风格：本地 MinIO 用 path-style（默认 true）；阿里云 OSS 等云形态用
    // virtual-host（S3_FORCE_PATH_STYLE=false + 基础 endpoint 不含 bucket）——
    // G1 实测：OSS 兼容口 path-style 会 403 签名不匹配。
    this.client = new S3Client({
      endpoint: config.endpoint,
      region: config.region,
      credentials: { accessKeyId: config.accessKey, secretAccessKey: config.secretKey },
      forcePathStyle: config.forcePathStyle ?? true,
    });
  }

  /**
   * 预签名上传。SigV4 预签名无法对 Content-Length 签名（规范排除）；所装版本的
   * s3-request-presigner 亦将 Content-Type 列为不可签名头（prepareRequest 强制
   * unsignableHeaders.add("content-type")，实测不带头的 PUT 返回 200）。
   * 因此返回的 headers 是「约定必须发送」而非签名强制：字节完整性由 verify 阶段
   * 的 head 核验，Content-Type 一致性由 complete 时服务端比对 head().contentType
   * 与会话申明值兜底（见 Task 19）。bytes 参数仅作为端口契约保留。
   */
  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; contentType: string } | null> {
    try {
      const out = await this.client.send(new HeadObjectCommand({ Bucket: this.bucket, Key: objectKey }));
      return { bytes: out.ContentLength ?? 0, etag: out.ETag ?? '', contentType: out.ContentType ?? '' };
    } 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 };
  }
}