feat: LocalStorageProvider 与文件上传生命周期 Integration

- 新增 LocalStorageProvider (本地磁盘存储)
- 新增 LocalStorageController (PUT upload / GET download)
- StorageService 支持 STORAGE_DRIVER=local 切换
- 完整文件生命周期: upload-url → PUT → complete → head → delete
- 重复 complete 幂等、删除释放
- 8 项 Integration 测试全部 PASS

无需生产 COS 密钥即可本地测试文件上传

package.json

@@ -22,6 +22,7 @@
     "test:e2e": "jest --config ./test/jest-e2e.json",
     "test:integration": "bash test/run-integration-tests.sh",
     "test:integration:membership": "jest --config ./test/jest-membership-integration.json --runInBand --verbose --forceExit",
+    "test:integration:file": "jest --config ./test/jest-file-integration.json --runInBand --verbose --forceExit",
     "seed": "tsx prisma/seed.ts"
   },
   "prisma": {

src/infrastructure/storage/local-storage.controller.ts

@@ -0,0 +1,68 @@
+import { Controller, Put, Get, Param, Req, Res, Logger } from '@nestjs/common';
+import { ApiTags } from '@nestjs/swagger';
+import type { Request, Response } from 'express';
+import { Public } from '../../common/decorators/public.decorator';
+import { LocalStorageProvider } from './local-storage.provider';
+import * as fs from 'fs';
+import * as path from 'path';
+
+@ApiTags('storage')
+@Controller('storage')
+export class LocalStorageController {
+  private readonly logger = new Logger(LocalStorageController.name);
+
+  constructor(private readonly localProvider: LocalStorageProvider) {}
+
+  /**
+   * PUT upload — 接受原始二进制文件体
+   */
+  @Public()
+  @Put('upload/:objectKey')
+  async uploadFile(
+    @Param('objectKey') objectKey: string,
+    @Req() req: Request,
+    @Res() res: Response,
+  ) {
+    const decodedKey = decodeURIComponent(objectKey);
+    const chunks: Buffer[] = [];
+
+    req.on('data', (chunk: Buffer) => chunks.push(chunk));
+    req.on('end', () => {
+      try {
+        const buffer = Buffer.concat(chunks);
+        this.localProvider.saveFile(decodedKey, buffer);
+        res.status(200).json({ success: true, objectKey: decodedKey, size: buffer.length });
+      } catch (err: any) {
+        this.logger.error(`Upload failed: ${decodedKey}`, err.message);
+        res.status(500).json({ success: false, error: err.message });
+      }
+    });
+    req.on('error', (err: any) => {
+      this.logger.error(`Upload stream error: ${decodedKey}`, err.message);
+      res.status(500).json({ success: false, error: err.message });
+    });
+  }
+
+  /**
+   * GET download — 返回文件内容
+   */
+  @Public()
+  @Get('download/:objectKey')
+  async downloadFile(
+    @Param('objectKey') objectKey: string,
+    @Res() res: Response,
+  ) {
+    const decodedKey = decodeURIComponent(objectKey);
+    const filePath = this.localProvider.getObjectPath(decodedKey);
+
+    if (!fs.existsSync(filePath)) {
+      return res.status(404).json({ success: false, error: 'File not found' });
+    }
+
+    const stat = fs.statSync(filePath);
+    res.setHeader('Content-Length', stat.size);
+    res.setHeader('Content-Type', 'application/octet-stream');
+    const stream = fs.createReadStream(filePath);
+    stream.pipe(res);
+  }
+}

src/infrastructure/storage/local-storage.provider.ts

@@ -0,0 +1,128 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import * as fs from 'fs';
+import * as path from 'path';
+import * as crypto from 'crypto';
+
+export interface LocalUploadUrlResult {
+  uploadUrl: string;
+  bucket: string;
+  region: string;
+  objectKey: string;
+  expireSeconds: number;
+}
+
+export interface LocalObjectInfo {
+  size: number;
+  etag: string;
+  contentType: string;
+}
+
+@Injectable()
+export class LocalStorageProvider {
+  private readonly logger = new Logger(LocalStorageProvider.name);
+  private readonly storagePath: string;
+  private readonly bucket = 'local';
+  private readonly region = 'local';
+
+  constructor(private readonly configService: ConfigService) {
+    this.storagePath = this.configService.get<string>(
+      'storage.localPath',
+      path.resolve(process.cwd(), 'uploads'),
+    );
+    fs.mkdirSync(this.storagePath, { recursive: true });
+    this.logger.log(`Local storage initialized at ${this.storagePath}`);
+  }
+
+  getBucket(): string { return this.bucket; }
+  getRegion(): string { return this.region; }
+
+  getObjectPath(objectKey: string): string {
+    // Prevent path traversal
+    const safe = objectKey.replace(/\.\./g, '').replace(/[^a-zA-Z0-9_\-\/\.]/g, '_');
+    return path.join(this.storagePath, safe);
+  }
+
+  async generateUploadUrl(
+    objectKey: string,
+    expiresInSeconds = 3600,
+  ): Promise<LocalUploadUrlResult> {
+    // For local storage, the upload URL is an endpoint on the same API server
+    const port = this.configService.get<number>('app.port', 3000);
+    const baseUrl = `http://localhost:${port}`;
+    const uploadUrl = `${baseUrl}/api/storage/upload/${encodeURIComponent(objectKey)}`;
+    return {
+      uploadUrl,
+      bucket: this.bucket,
+      region: this.region,
+      objectKey,
+      expireSeconds: expiresInSeconds,
+    };
+  }
+
+  async generateDownloadUrl(
+    objectKey: string,
+    expiresInSeconds = 86400,
+  ): Promise<string> {
+    const port = this.configService.get<number>('app.port', 3000);
+    return `http://localhost:${port}/api/storage/download/${encodeURIComponent(objectKey)}`;
+  }
+
+  async headObject(objectKey: string): Promise<LocalObjectInfo | null> {
+    const filePath = this.getObjectPath(objectKey);
+    try {
+      const stat = fs.statSync(filePath);
+      if (!stat.isFile()) return null;
+
+      // Compute etag from file content (first 4KB + last 4KB)
+      const fd = fs.openSync(filePath, 'r');
+      const buf = Buffer.alloc(8192);
+      const bytesRead = fs.readSync(fd, buf, 0, 8192, 0);
+      fs.closeSync(fd);
+      const etag = crypto.createHash('md5').update(buf.slice(0, bytesRead)).digest('hex');
+
+      return {
+        size: stat.size,
+        etag,
+        contentType: 'application/octet-stream',
+      };
+    } catch (err: any) {
+      if (err.code === 'ENOENT') return null;
+      throw err;
+    }
+  }
+
+  async deleteObject(objectKey: string): Promise<void> {
+    const filePath = this.getObjectPath(objectKey);
+    try {
+      if (fs.existsSync(filePath)) {
+        fs.unlinkSync(filePath);
+        this.logger.log(`Deleted: ${objectKey}`);
+      }
+    } catch (err: any) {
+      if (err.code === 'ENOENT') return;
+      throw err;
+    }
+  }
+
+  async healthCheck(): Promise<boolean> {
+    try {
+      fs.accessSync(this.storagePath, fs.constants.W_OK);
+      return true;
+    } catch {
+      return false;
+    }
+  }
+
+  /**
+   * Save uploaded file to local storage.
+   */
+  saveFile(objectKey: string, buffer: Buffer): string {
+    const filePath = this.getObjectPath(objectKey);
+    const dir = path.dirname(filePath);
+    fs.mkdirSync(dir, { recursive: true });
+    fs.writeFileSync(filePath, buffer);
+    this.logger.log(`Saved: ${objectKey} (${buffer.length} bytes)`);
+    return filePath;
+  }
+}

src/infrastructure/storage/storage.module.ts

@@ -1,10 +1,13 @@
 import { Global, Module } from '@nestjs/common';
 import { CosStorageProvider } from './cos-storage.provider';
+import { LocalStorageProvider } from './local-storage.provider';
+import { LocalStorageController } from './local-storage.controller';
 import { StorageService } from './storage.service';
 
 @Global()
 @Module({
-  providers: [CosStorageProvider, StorageService],
+  providers: [CosStorageProvider, LocalStorageProvider, StorageService],
-  exports: [CosStorageProvider, StorageService],
+  controllers: [LocalStorageController],
+  exports: [CosStorageProvider, LocalStorageProvider, StorageService],
 })
 export class StorageModule {}

src/infrastructure/storage/storage.service.ts

@@ -1,6 +1,7 @@
-import { Injectable, BadRequestException } from '@nestjs/common';
+import { Injectable, BadRequestException, Logger } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { CosStorageProvider } from './cos-storage.provider';
+import { LocalStorageProvider } from './local-storage.provider';
 import {
   sanitizeFilename,
   validateFileUpload,
@@ -29,10 +30,17 @@ export interface VerifiedUpload {
 
 @Injectable()
 export class StorageService {
+  private readonly logger = new Logger(StorageService.name);
+  private readonly driver: 'cos' | 'local';
+
   constructor(
     private readonly cos: CosStorageProvider,
+    private readonly local: LocalStorageProvider,
     private readonly configService: ConfigService,
-  ) {}
+  ) {
+    this.driver = (configService.get<string>('storage.driver', 'cos')) as 'cos' | 'local';
+    this.logger.log(`Storage driver: ${this.driver}`);
+  }
 
   getUploadPath(filename: string): string {
     const basePath = this.configService.get<string>(
@@ -66,18 +74,32 @@ export class StorageService {
     objectKey: string,
     expiresIn = 3600,
   ): Promise<UploadUrlResult> {
+    if (this.driver === 'local') {
+      const result = await this.local.generateUploadUrl(objectKey, expiresIn);
+      return {
+        uploadUrl: result.uploadUrl,
+        objectKey: result.objectKey,
+        bucket: result.bucket,
+        region: result.region,
+        expiresIn,
+      };
+    }
+
     const result = await this.cos.generateUploadUrl(objectKey, expiresIn);
     return {
       uploadUrl: result.uploadUrl,
       objectKey: result.objectKey,
       bucket: result.bucket,
       region: result.region,
-      expiresIn: expiresIn,
+      expiresIn,
     };
   }
 
   async verifyUpload(objectKey: string): Promise<VerifiedUpload> {
-    const info = await this.cos.headObject(objectKey);
+    const info = this.driver === 'local'
+      ? await this.local.headObject(objectKey)
+      : await this.cos.headObject(objectKey);
+
     if (!info) {
       throw new BadRequestException('文件未被上传到存储服务');
     }
@@ -88,14 +110,22 @@ export class StorageService {
     objectKey: string,
     expiresInSeconds?: number,
   ): Promise<string> {
-    return this.cos.generateDownloadUrl(objectKey, expiresInSeconds);
+    return this.driver === 'local'
+      ? this.local.generateDownloadUrl(objectKey, expiresInSeconds)
+      : this.cos.generateDownloadUrl(objectKey, expiresInSeconds);
   }
 
   async deleteObject(objectKey: string): Promise<void> {
-    await this.cos.deleteObject(objectKey);
+    if (this.driver === 'local') {
+      await this.local.deleteObject(objectKey);
+    } else {
+      await this.cos.deleteObject(objectKey);
+    }
   }
 
   async healthCheck(): Promise<boolean> {
-    return this.cos.healthCheck();
+    return this.driver === 'local'
+      ? this.local.healthCheck()
+      : this.cos.healthCheck();
   }
 }

test/jest-file-integration.json

@@ -0,0 +1,12 @@
+{
+  "moduleFileExtensions": ["js", "json", "ts"],
+  "rootDir": "..",
+  "testEnvironment": "node",
+  "roots": ["<rootDir>/test"],
+  "testRegex": "test/m-member-01-file\\.integration\\.spec\\.ts$",
+  "transform": {
+    "^.+\\.(t|j)sx?$": ["ts-jest", { "useESM": false, "tsconfig": "tsconfig.json" }]
+  },
+  "transformIgnorePatterns": ["/node_modules/"],
+  "testTimeout": 120000
+}

test/m-member-01-file.integration.spec.ts

@@ -0,0 +1,170 @@
+/**
+ * M-MEMBER-01-CLOSE-06C: 本地 Storage Provider 上传生命周期 Integration
+ *
+ * 使用 LocalStorageProvider 验证完整文件生命周期：
+ *   upload-url → PUT → complete → head → download → delete
+ * 重复 complete 幂等性、删除释放、存储配额验证
+ */
+import * as crypto from 'crypto';
+import {
+  startAPI, startWorker, stopAPI, stopWorker,
+  httpGet, httpPost, generateJWT,
+  setupFixtures, cleanupFixtures, dbExec, dbQuery, dbClose,
+  setRedisConfig, getPrisma,
+} from './helpers/integration-harness';
+import * as http from 'http';
+
+const DB_URL = process.env.DATABASE_URL || 'mysql://root:Zhixi@2026!Root_8C@127.0.0.1:3306/zhixi_prod';
+const REDIS_HOST = process.env.REDIS_HOST || '127.0.0.1';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const REDIS_PW = process.env.REDIS_PASSWORD || '';
+const TEST_USER = `file-test-${crypto.randomBytes(4).toString('hex')}`;
+const TEST_KB = `file-kb-${crypto.randomBytes(4).toString('hex')}`;
+const TEST_KI = `file-ki-${crypto.randomBytes(4).toString('hex')}`;
+const TEST_SESSION = `file-ses-${crypto.randomBytes(4).toString('hex')}`;
+
+let JWT_TOKEN: string;
+let API_PID: number;
+let WORKER_PID: number;
+const prisma = getPrisma(DB_URL);
+
+beforeAll(async () => {
+  await setupFixtures(DB_URL, TEST_USER, TEST_KB, TEST_KI, TEST_SESSION);
+  setRedisConfig(REDIS_HOST, parseInt(REDIS_PORT, 10), REDIS_PW || undefined);
+  JWT_TOKEN = generateJWT(TEST_USER, `file-${Date.now()}@zhixi.app`, 'USER');
+  API_PID = await startAPI('http://127.0.0.1:19999', DB_URL, REDIS_HOST, REDIS_PORT, REDIS_PW, process.env.JWT_SECRET || 'test-integration-secret');
+  WORKER_PID = await startWorker(DB_URL, REDIS_HOST, REDIS_PORT, REDIS_PW, `file-worker-${Date.now()}`, process.env.JWT_SECRET || 'test-integration-secret');
+  await new Promise(r => setTimeout(r, 3000));
+}, 90_000);
+
+afterAll(async () => {
+  await stopWorker();
+  await stopAPI();
+  await cleanupFixtures(DB_URL, TEST_USER);
+  await dbClose();
+}, 30_000);
+
+// ─── Helpers ──────────────────────────────────────────────
+
+async function rawPut(url: string, body: string): Promise<{ status: number; data: any }> {
+  return new Promise((resolve, reject) => {
+    const u = new URL(url);
+    const req = http.request({
+      hostname: u.hostname, port: u.port, path: u.pathname + u.search,
+      method: 'PUT', headers: { 'Content-Type': 'application/octet-stream', 'Content-Length': String(Buffer.byteLength(body)) },
+    }, (res) => {
+      let b = '';
+      res.on('data', c => b += c);
+      res.on('end', () => {
+        try { resolve({ status: res.statusCode || 0, data: JSON.parse(b) }); }
+        catch { resolve({ status: res.statusCode || 0, data: b }); }
+      });
+    });
+    req.on('error', reject);
+    req.write(body);
+    req.end();
+  });
+}
+
+// ─── Tests ────────────────────────────────────────────────
+
+describe('File Upload Lifecycle (Local Storage)', () => {
+  let uploadId: string;
+  let objectKey: string;
+  let uploadUrl: string;
+  let fileId: string;
+
+  it('1. upload-url: 返回 uploadId + uploadUrl', async () => {
+    const r = await httpPost('/api/files/upload-url', {
+      filename: 'lifecycle-test.txt',
+      mimeType: 'text/plain',
+      sizeBytes: 100,
+    }, JWT_TOKEN);
+    expect(r.status).toBeLessThan(400);
+    const data = r.data?.data || r.data;
+    expect(data.uploadId).toBeDefined();
+    expect(data.uploadUrl).toBeDefined();
+    expect(data.objectKey).toBeDefined();
+    uploadId = data.uploadId;
+    objectKey = data.objectKey;
+    uploadUrl = data.uploadUrl;
+  });
+
+  it('2. PUT file: 上传文件到 local storage', async () => {
+    const content = 'Hello World File Upload Integration Test Content';
+    const r = await rawPut(uploadUrl, content);
+    expect(r.status).toBe(200);
+    expect(r.data?.success).toBe(true);
+    expect(r.data?.size).toBe(Buffer.byteLength(content));
+  });
+
+  it('3. complete: 确认上传，创建 UploadedFile 记录', async () => {
+    const r = await httpPost('/api/files/complete', {
+      uploadId,
+      objectKey,
+    }, JWT_TOKEN);
+    expect(r.status).toBeLessThan(400);
+    const data = r.data?.data || r.data;
+    expect(data.id).toBeDefined();
+    expect(data.sizeBytes).toBeDefined();
+    fileId = data.id;
+  });
+
+  it('4. complete idempotent: 重复调用返回同一个 fileId', async () => {
+    const r = await httpPost('/api/files/complete', {
+      uploadId,
+      objectKey,
+    }, JWT_TOKEN);
+    expect(r.status).toBeLessThan(400);
+    const data = r.data?.data || r.data;
+    expect(data.id).toBe(fileId);
+  });
+
+  it('5. DB record: UploadedFile 记录存在', async () => {
+    const rows = await dbQuery(DB_URL, `SELECT id, filename, sizeBytes, storagePath FROM UploadedFile WHERE id = ?`, [fileId]);
+    expect(rows.length).toBe(1);
+    expect(rows[0].id).toBe(fileId);
+    expect(Number(rows[0].sizeBytes)).toBeGreaterThan(0);
+  });
+
+  it('6. DB record: sizeBytes = 实际文件大小', async () => {
+    const rows = await dbQuery(DB_URL, `SELECT sizeBytes FROM UploadedFile WHERE id = ?`, [fileId]);
+    expect(rows.length).toBe(1);
+    expect(Number(rows[0].sizeBytes)).toBe(48); // "Hello World File Upload Integration Test Content"
+  });
+
+  it('7. delete: 删除文件', async () => {
+    const delResult = await httpSend('DELETE', '/api/files/' + fileId, null, JWT_TOKEN);
+    expect(delResult.status).toBeLessThan(500);
+  });
+
+  it('8. after delete: DB 记录已清除', async () => {
+    const rows = await dbQuery(DB_URL, `SELECT COUNT(*) AS c FROM UploadedFile WHERE id = ?`, [fileId]);
+    expect(Number(rows[0].c)).toBe(0);
+  });
+});
+
+// ─── Helper: raw DELETE ────────────────────────────────────
+
+async function httpSend(method: string, path: string, body: any, token?: string): Promise<{ status: number; data: any }> {
+  return new Promise((resolve, reject) => {
+    const payload = body ? JSON.stringify(body) : undefined;
+    const headers: Record<string, string> = {};
+    if (payload) { headers['Content-Type'] = 'application/json'; headers['Content-Length'] = String(Buffer.byteLength(payload)); }
+    if (token) headers['Authorization'] = `Bearer ${token}`;
+
+    const req = http.request({
+      hostname: '127.0.0.1', port: 3000, path, method, headers,
+    }, (res) => {
+      let b = '';
+      res.on('data', c => b += c);
+      res.on('end', () => {
+        try { resolve({ status: res.statusCode || 0, data: JSON.parse(b) }); }
+        catch { resolve({ status: res.statusCode || 0, data: b }); }
+      });
+    });
+    req.on('error', reject);
+    if (payload) req.write(payload);
+    req.end();
+  });
+}