16.10-文件存储服务

要点

业务逻辑不应该直接依赖 R2 或 S3 的 API——加一层 StorageProvider 接口抽象，R2 和 S3 两种实现共享同一套签名
文件元数据（名称、大小、类型、上传者、处理状态）存在 D1，和对象存储分离
访问控制分三档：公开文件走 CDN、私有文件走 Worker 代理鉴权、临时分享链接带过期时间
文件生命周期：pending → processing → completed / failed → deleted，配合定时任务做清理
通过 Hono 中间件注入存储实现，环境变量决定用哪个后端

1. 为什么需要存储抽象层

前面的文章里，上传文件直接调用 c.env.BUCKET.put(key, body)，下载直接调用 bucket.get(key)。如果项目只用 R2、只跑在 Cloudflare 上，这样做没问题。

但实际情况往往会有变化：本地开发没有 R2 绑定、团队同时维护 Cloudflare 和 AWS 两套部署、自建环境用 MinIO、测试时需要 mock 存储层。这些场景都指向同一件事：业务代码应该依赖一个接口，而不是某个具体的存储 SDK。

在 Workers 环境里做这件事有一个方便的条件——存储操作的接口足够简单。上传、下载、删除、列出，四个操作就能覆盖大部分需求，用 TypeScript 的 interface 就能把契约描述清楚。

2. 统一的 StorageProvider 接口

// src/storage/types.ts
export interface StorageProvider {
  put(
    key: string,
    data: Uint8Array | ReadableStream,
    options: { contentType?: string; metadata?: Record<string, string> }
  ): Promise<void>
 
  get(key: string): Promise<{
    data: ReadableStream
    contentType?: string
    metadata?: Record<string, string>
  } | null>
 
  delete(key: string): Promise<void>
 
  list(prefix: string, limit?: number): Promise<
    Array<{ key: string; size: number; uploaded: Date }>
  >
}

几个设计考虑：

data 用 Uint8Array | ReadableStream——R2 和 S3 都支持这两种格式，接口层不需要做转换
metadata 是业务元数据——上传者 ID、关联文档 ID 这类信息。在 R2 里对应 customMetadata，在 S3 里对应 Metadata
接口只描述文件操作，不管元数据持久化——文件记录存 D1 还是 PostgreSQL，跟这个接口无关

3. R2 实现

R2 和 Workers 原生集成，通过 R2Bucket 绑定直接访问，不需要额外的认证配置。

// src/storage/r2-provider.ts
import type { StorageProvider } from './types'
 
export class R2StorageProvider implements StorageProvider {
  constructor(private bucket: R2Bucket) {}
 
  async put(key: string, data: Uint8Array | ReadableStream, options: {
    contentType?: string; metadata?: Record<string, string>
  }): Promise<void> {
    await this.bucket.put(key, data, {
      httpMetadata: { contentType: options.contentType ?? 'application/octet-stream' },
      customMetadata: options.metadata,
    })
  }
 
  async get(key: string) {
    const object = await this.bucket.get(key)
    if (!object) return null
    return {
      data: object.body,
      contentType: object.httpMetadata?.contentType,
      metadata: object.customMetadata,
    }
  }
 
  async delete(key: string): Promise<void> {
    await this.bucket.delete(key)
  }
 
  async list(prefix: string, limit = 1000) {
    const listed = await this.bucket.list({ prefix, limit })
    return listed.objects.map((obj) => ({
      key: obj.key, size: obj.size, uploaded: obj.uploaded,
    }))
  }
}

R2 的优势：零出口流量费（S3 收 $0.09/GB）、和 Workers 深度集成不需要配置 access key、S3 兼容 API 让外部工具也能操作同一个 bucket。

4. S3 兼容实现

AWS S3、MinIO、Backblaze B2 都支持 S3 API。用 @aws-sdk/client-s3 可以覆盖这些平台：

// src/storage/s3-provider.ts
import { S3Client, PutObjectCommand, GetObjectCommand,
         DeleteObjectCommand, ListObjectsV2Command } from '@aws-sdk/client-s3'
import type { StorageProvider } from './types'
 
export class S3StorageProvider implements StorageProvider {
  constructor(private bucket: string, private client: S3Client) {}
 
  async put(key: string, data: Uint8Array | ReadableStream, options: {
    contentType?: string; metadata?: Record<string, string>
  }): Promise<void> {
    await this.client.send(new PutObjectCommand({
      Bucket: this.bucket, Key: key, Body: data,
      ContentType: options.contentType, Metadata: options.metadata,
    }))
  }
 
  async get(key: string) {
    const output = await this.client.send(new GetObjectCommand({
      Bucket: this.bucket, Key: key,
    }))
    if (!output.Body) return null
    return {
      data: output.Body as ReadableStream,
      contentType: output.ContentType,
      metadata: output.Metadata,
    }
  }
 
  async delete(key: string): Promise<void> {
    await this.client.send(new DeleteObjectCommand({ Bucket: this.bucket, Key: key }))
  }
 
  async list(prefix: string, limit = 1000) {
    const output = await this.client.send(new ListObjectsV2Command({
      Bucket: this.bucket, Prefix: prefix, MaxKeys: limit,
    }))
    return (output.Contents ?? []).map((obj) => ({
      key: obj.Key!, size: obj.Size ?? 0, uploaded: obj.LastModified ?? new Date(),
    }))
  }
}
 
export function createS3Provider(options: {
  endpoint?: string; region: string
  accessKeyId: string; secretAccessKey: string; bucket: string
}): S3StorageProvider {
  const client = new S3Client({
    endpoint: options.endpoint, region: options.region,
    credentials: { accessKeyId: options.accessKeyId, secretAccessKey: options.secretAccessKey },
    forcePathStyle: !!options.endpoint, // MinIO 等自建服务需要
  })
  return new S3StorageProvider(options.bucket, client)
}

两种实现的对比：R2 零出口流量费、Worker 绑定直接调用，但签名 URL 需要 Worker 代理；S3 兼容 API 覆盖 AWS / MinIO / 多云场景，SDK 内置 presigned URL，但需要配置认证。

在 Cloudflare 上跑的项目，R2 是更合适的默认选择。需要多云部署或者已有 S3 资产的场景，再切 S3 实现。

5. 文件元数据管理

对象存储只管文件内容和简单的 HTTP 头信息。业务层面的元数据——原始文件名、上传者、处理状态、关联的业务实体——存在 D1 里。

表结构

// src/db/schema.ts
import { sqliteTable, text, integer } from 'drizzle-orm/sqlite-core'
 
export const files = sqliteTable('files', {
  id: text('id').primaryKey(),
  key: text('key').notNull().unique(),            // 对象存储的 key
  originalName: text('original_name').notNull(),  // 用户上传时的文件名
  mimeType: text('mime_type').notNull(),
  size: integer('size').notNull(),
  status: text('status').notNull().default('pending'),
  userId: text('user_id').notNull(),
  tenantId: text('tenant_id').notNull(),
  knowledgeBaseId: text('knowledge_base_id'),     // 关联的知识库
  processingResult: text('processing_result'),    // 处理结果（如解析后的文本）
  errorMessage: text('error_message'),
  createdAt: integer('created_at', { mode: 'timestamp' }).notNull(),
  updatedAt: integer('updated_at', { mode: 'timestamp' }).notNull(),
})
 
export type FileStatus =
  | 'pending'     // 已上传，等待处理
  | 'processing'  // 处理中
  | 'completed'   // 处理完成
  | 'failed'      // 处理失败
  | 'deleted'     // 已标记删除（软删除）

对象存储和数据库的分工：对象存储负责文件二进制内容 + Content-Type + 轻量自定义 metadata；D1 负责业务元数据 + 状态流转 + 权限关联 + 处理结果。

记录操作

async function createFileRecord(
  db: DrizzleD1Database,
  input: { key: string; originalName: string; mimeType: string
           size: number; userId: string; tenantId: string }
) {
  const record = {
    id: crypto.randomUUID(), ...input,
    status: 'pending' as const,
    createdAt: new Date(), updatedAt: new Date(),
  }
  await db.insert(files).values(record)
  return record
}
 
async function updateFileStatus(
  db: DrizzleD1Database, fileId: string, status: FileStatus,
  extra?: { processingResult?: string; errorMessage?: string }
) {
  await db.update(files)
    .set({ status, ...extra, updatedAt: new Date() })
    .where(eq(files.id, fileId))
}

6. 访问控制

访问控制分三种场景。

公开文件

头像、公开文档附件——任何人拿到 URL 就能访问，直接走 CDN，不经过 Worker：

function getPublicUrl(key: string, publicDomain: string): string {
  return `https://${publicDomain}/${key}`
}

私有文件

合同、个人数据——只有鉴权用户能访问。Worker 充当下载代理，先查权限再返回文件：

app.get('/api/files/:key/download', async (c) => {
  const user = c.get('user')
  const storage = c.get('storage') as StorageProvider
  const key = decodeURIComponent(c.req.param('key'))
  const record = await db.query.files.findFirst({ where: eq(files.key, key) })
  if (!record) return c.json({ error: 'Not found' }, 404)
  if (record.tenantId !== user.tenantId) return c.json({ error: 'Forbidden' }, 403)
 
  const result = await storage.get(key)
  if (!result) return c.json({ error: 'File not found' }, 404)
  return new Response(result.data, {
    headers: {
      'Content-Type': result.contentType ?? 'application/octet-stream',
      'Content-Disposition': `inline; filename="${record.originalName}"`,
    },
  })
})

代价是每次下载都经过 Worker，下载量大时 CPU 时间和请求数都会增加。好处是权限检查完全可控，不存在签名 URL 泄露后被滥用的风险。

临时分享链接

用户想把私有文件分享给外部人员，生成带过期时间的 token：

app.post('/api/files/:id/share', async (c) => {
  const user = c.get('user')
  const { expiresInHours = 24 } = await c.req.json()
  const token = crypto.randomUUID()
  const expiresAt = new Date(Date.now() + expiresInHours * 3600 * 1000)
 
  await db.insert(shareLinks).values({
    id: crypto.randomUUID(), token, fileId: c.req.param('id'),
    createdBy: user.id, expiresAt, usedCount: 0, createdAt: new Date(),
  })
  return c.json({ shareUrl: `https://example.com/share/${token}`, expiresAt })
})
 
app.get('/share/:token', async (c) => {
  const storage = c.get('storage') as StorageProvider
  const link = await db.query.shareLinks.findFirst({
    where: eq(shareLinks.token, c.req.param('token')),
  })
  if (!link) return c.json({ error: 'Not found' }, 404)
  if (link.expiresAt < new Date()) return c.json({ error: 'Expired' }, 410)
 
  const record = await db.query.files.findFirst({ where: eq(files.id, link.fileId) })
  if (!record) return c.json({ error: 'File not found' }, 404)
  const result = await storage.get(record.key)
  if (!result) return c.json({ error: 'File not found' }, 404)
 
  await db.update(shareLinks).set({ usedCount: link.usedCount + 1 })
    .where(eq(shareLinks.id, link.id))
 
  return new Response(result.data, {
    headers: { 'Content-Type': record.mimeType,
      'Content-Disposition': `inline; filename="${record.originalName}"` },
  })
})

7. 文件生命周期

文件从上传到删除，经历一组明确的状态：

pending → processing → completed → deleted
                  ↘ failed → processing（重试）

上传和处理

async function uploadAndQueue(c: Context, file: File, userId: string, tenantId: string) {
  const storage = c.get('storage') as StorageProvider
  const key = `${tenantId}/${crypto.randomUUID()}/${file.name}`
 
  await storage.put(key, new Uint8Array(await file.arrayBuffer()), {
    contentType: file.type, metadata: { userId, originalName: file.name },
  })
  const record = await createFileRecord(c.get('db'), {
    key, originalName: file.name, mimeType: file.type,
    size: file.size, userId, tenantId,
  })
  await c.env.FILE_PROCESSING_QUEUE.push({
    fileId: record.id, key: record.key, mimeType: record.mimeType,
  })
  return record
}

消费者从队列取出任务后，根据 mimeType 分发到不同处理器（PDF 解析、OCR、纯文本提取等），处理完更新状态为 completed，失败则更新为 failed 并记录 errorMessage。具体的队列消费模式参考工作流章节。

清理策略

用户删除文件时做软删除，实际清理交给 Cron Trigger 定时任务：

async function runCleanupJobs(storage: StorageProvider) {
  const now = new Date()
  const daysAgo = (n: number) => new Date(now.getTime() - n * 86400_000)
 
  // 失败超过 7 天的文件——给用户留了重试窗口
  for (const f of await db.query.files.findMany({
    where: and(eq(files.status, 'failed'), lt(files.updatedAt, daysAgo(7))),
  })) {
    await storage.delete(f.key)
    await db.delete(files).where(eq(files.id, f.id))
  }
 
  // 软删除超过 30 天的文件——回收站到期
  for (const f of await db.query.files.findMany({
    where: and(eq(files.status, 'deleted'), lt(files.updatedAt, daysAgo(30))),
  })) {
    await storage.delete(f.key)
    await db.delete(files).where(eq(files.id, f.id))
  }
 
  // 临时文件（tmp/ 前缀）超过 24 小时
  const oneDayAgo = daysAgo(1)
  for (const f of await storage.list('tmp/')) {
    if (f.uploaded < oneDayAgo) await storage.delete(f.key)
  }
}

8. 存储成本优化

对象存储的成本来自三个方面：存储容量、出口流量、请求次数。

过期策略是最直接的手段。不同类型的数据设置不同的生命周期：临时文件 24 小时、处理失败文件 7 天、软删除文件 30 天、正式文件跟随业务需要长期保留。

出口流量在 S3 上是 $0.09/GB，累积起来不是小数。R2 免出口流量费，部署在 Cloudflare 上的项目迁移到 R2 可以直接省掉这笔开支。

存储分层——冷数据（很少访问的归档文件）可以迁移到低频存储。S3 有 Glacier，R2 暂时没有分层，但可以通过定期扫描数据库里长期未访问的文件，把它们移到单独的 bucket 或导出到外部归档。

存储用量查询——给用户展示配额，也可以用来生成账单：

async function getStorageStats(tenantId: string) {
  const result = await db
    .select({ totalFiles: count(files.id), totalSize: sum(files.size) })
    .from(files)
    .where(and(eq(files.tenantId, tenantId), ne(files.status, 'deleted')))
  return result[0]
}

9. 依赖注入：切换存储后端

在应用启动时根据环境变量创建对应的 StorageProvider，通过 Hono 中间件注入到请求上下文：

// src/storage/factory.ts
export function createStorageProvider(env: Env): StorageProvider {
  switch (env.STORAGE_BACKEND) {
    case 'r2':
      return new R2StorageProvider(env.BUCKET)
    case 's3':
      return createS3Provider({
        endpoint: env.S3_ENDPOINT, region: env.S3_REGION,
        accessKeyId: env.S3_ACCESS_KEY, secretAccessKey: env.S3_SECRET_KEY,
        bucket: env.S3_BUCKET,
      })
    case 'memory':
      return new InMemoryStorageProvider() // 本地开发用，用 Map 模拟
    default:
      throw new Error(`Unknown storage backend: ${env.STORAGE_BACKEND}`)
  }
}

// src/index.ts
import { createStorageProvider } from './storage/factory'
 
app.use('*', async (c, next) => {
  c.set('storage', createStorageProvider(c.env))
  await next()
})

InMemoryStorageProvider 用一个 Map<string, { data, contentType, metadata, uploaded }> 模拟存储操作，本地开发时不需要任何外部服务。.dev.vars 里设置 STORAGE_BACKEND=memory，部署到 Cloudflare 时设置为 r2，切换到 AWS 时设置为 s3 并补上对应环境变量。业务代码全程只调用 StorageProvider 接口的方法。

总结

回顾这篇的要点：

存储抽象层让业务代码不绑死某个具体的存储 SDK，切换后端、本地开发、测试 mock 都更方便
StorageProvider 接口定义四个核心操作——R2 和 S3 两种实现共享同一套签名
文件元数据存在 D1，和对象存储分离，状态流转在数据库里管理
访问控制分三档：公开文件走 CDN、私有文件走 Worker 代理鉴权、临时链接带过期时间
文件生命周期有明确的状态机：pending → processing → completed / failed → deleted
清理策略用定时任务执行——失败文件保留 7 天，软删除保留 30 天，临时文件 24 小时
成本优化：过期策略、R2 零出口流量、存储统计查配额
依赖注入通过工厂函数 + 中间件实现，环境变量决定用哪个后端

下一篇讲对象存储接入的具体细节——R2 的高级用法和 S3 兼容 API 的完整接入流程。