feat(file): unify file handling with a new FileSource abstraction for URL and base64 data

This commit is contained in:
CaIon
2026-02-04 17:15:24 +08:00
parent 89b2782675
commit 9ef7740fe7
28 changed files with 1119 additions and 357 deletions

View File

@@ -5,12 +5,9 @@ import (
"fmt"
"io"
"os"
"path/filepath"
"sync"
"sync/atomic"
"time"
"github.com/google/uuid"
)
// BodyStorage 请求体存储接口
@@ -101,25 +98,10 @@ type diskStorage struct {
}
func newDiskStorage(data []byte, cachePath string) (*diskStorage, error) {
// 确定缓存目录
dir := cachePath
if dir == "" {
dir = os.TempDir()
}
dir = filepath.Join(dir, "new-api-body-cache")
// 确保目录存在
if err := os.MkdirAll(dir, 0755); err != nil {
return nil, fmt.Errorf("failed to create cache directory: %w", err)
}
// 创建临时文件
filename := fmt.Sprintf("body-%s-%d.tmp", uuid.New().String()[:8], time.Now().UnixNano())
filePath := filepath.Join(dir, filename)
file, err := os.OpenFile(filePath, os.O_CREATE|os.O_RDWR|os.O_EXCL, 0600)
// 使用统一的缓存目录管理
filePath, file, err := CreateDiskCacheFile(DiskCacheTypeBody)
if err != nil {
return nil, fmt.Errorf("failed to create temp file: %w", err)
return nil, err
}
// 写入数据
@@ -148,25 +130,10 @@ func newDiskStorage(data []byte, cachePath string) (*diskStorage, error) {
}
func newDiskStorageFromReader(reader io.Reader, maxBytes int64, cachePath string) (*diskStorage, error) {
// 确定缓存目录
dir := cachePath
if dir == "" {
dir = os.TempDir()
}
dir = filepath.Join(dir, "new-api-body-cache")
// 确保目录存在
if err := os.MkdirAll(dir, 0755); err != nil {
return nil, fmt.Errorf("failed to create cache directory: %w", err)
}
// 创建临时文件
filename := fmt.Sprintf("body-%s-%d.tmp", uuid.New().String()[:8], time.Now().UnixNano())
filePath := filepath.Join(dir, filename)
file, err := os.OpenFile(filePath, os.O_CREATE|os.O_RDWR|os.O_EXCL, 0600)
// 使用统一的缓存目录管理
filePath, file, err := CreateDiskCacheFile(DiskCacheTypeBody)
if err != nil {
return nil, fmt.Errorf("failed to create temp file: %w", err)
return nil, err
}
// 从 reader 读取并写入文件
@@ -337,29 +304,6 @@ func CreateBodyStorageFromReader(reader io.Reader, contentLength int64, maxBytes
// CleanupOldCacheFiles 清理旧的缓存文件(用于启动时清理残留)
func CleanupOldCacheFiles() {
cachePath := GetDiskCachePath()
if cachePath == "" {
cachePath = os.TempDir()
}
dir := filepath.Join(cachePath, "new-api-body-cache")
entries, err := os.ReadDir(dir)
if err != nil {
return // 目录不存在或无法读取
}
now := time.Now()
for _, entry := range entries {
if entry.IsDir() {
continue
}
info, err := entry.Info()
if err != nil {
continue
}
// 删除超过 5 分钟的旧文件
if now.Sub(info.ModTime()) > 5*time.Minute {
os.Remove(filepath.Join(dir, entry.Name()))
}
}
// 使用统一的缓存管理
CleanupOldDiskCacheFiles(5 * time.Minute)
}

172
common/disk_cache.go Normal file
View File

@@ -0,0 +1,172 @@
package common
import (
"fmt"
"os"
"path/filepath"
"time"
"github.com/google/uuid"
)
// DiskCacheType 磁盘缓存类型
type DiskCacheType string
const (
DiskCacheTypeBody DiskCacheType = "body" // 请求体缓存
DiskCacheTypeFile DiskCacheType = "file" // 文件数据缓存
)
// 统一的缓存目录名
const diskCacheDir = "new-api-body-cache"
// GetDiskCacheDir 获取统一的磁盘缓存目录
// 注意:每次调用都会重新计算,以响应配置变化
func GetDiskCacheDir() string {
cachePath := GetDiskCachePath()
if cachePath == "" {
cachePath = os.TempDir()
}
return filepath.Join(cachePath, diskCacheDir)
}
// EnsureDiskCacheDir 确保缓存目录存在
func EnsureDiskCacheDir() error {
dir := GetDiskCacheDir()
return os.MkdirAll(dir, 0755)
}
// CreateDiskCacheFile 创建磁盘缓存文件
// cacheType: 缓存类型body/file
// 返回文件路径和文件句柄
func CreateDiskCacheFile(cacheType DiskCacheType) (string, *os.File, error) {
if err := EnsureDiskCacheDir(); err != nil {
return "", nil, fmt.Errorf("failed to create cache directory: %w", err)
}
dir := GetDiskCacheDir()
filename := fmt.Sprintf("%s-%s-%d.tmp", cacheType, uuid.New().String()[:8], time.Now().UnixNano())
filePath := filepath.Join(dir, filename)
file, err := os.OpenFile(filePath, os.O_CREATE|os.O_RDWR|os.O_EXCL, 0600)
if err != nil {
return "", nil, fmt.Errorf("failed to create cache file: %w", err)
}
return filePath, file, nil
}
// WriteDiskCacheFile 写入数据到磁盘缓存文件
// 返回文件路径
func WriteDiskCacheFile(cacheType DiskCacheType, data []byte) (string, error) {
filePath, file, err := CreateDiskCacheFile(cacheType)
if err != nil {
return "", err
}
_, err = file.Write(data)
if err != nil {
file.Close()
os.Remove(filePath)
return "", fmt.Errorf("failed to write cache file: %w", err)
}
if err := file.Close(); err != nil {
os.Remove(filePath)
return "", fmt.Errorf("failed to close cache file: %w", err)
}
return filePath, nil
}
// WriteDiskCacheFileString 写入字符串到磁盘缓存文件
func WriteDiskCacheFileString(cacheType DiskCacheType, data string) (string, error) {
return WriteDiskCacheFile(cacheType, []byte(data))
}
// ReadDiskCacheFile 读取磁盘缓存文件
func ReadDiskCacheFile(filePath string) ([]byte, error) {
return os.ReadFile(filePath)
}
// ReadDiskCacheFileString 读取磁盘缓存文件为字符串
func ReadDiskCacheFileString(filePath string) (string, error) {
data, err := os.ReadFile(filePath)
if err != nil {
return "", err
}
return string(data), nil
}
// RemoveDiskCacheFile 删除磁盘缓存文件
func RemoveDiskCacheFile(filePath string) error {
return os.Remove(filePath)
}
// CleanupOldDiskCacheFiles 清理旧的缓存文件
// maxAge: 文件最大存活时间
// 注意:此函数只删除文件,不更新统计(因为无法知道每个文件的原始大小)
func CleanupOldDiskCacheFiles(maxAge time.Duration) error {
dir := GetDiskCacheDir()
entries, err := os.ReadDir(dir)
if err != nil {
if os.IsNotExist(err) {
return nil // 目录不存在,无需清理
}
return err
}
now := time.Now()
for _, entry := range entries {
if entry.IsDir() {
continue
}
info, err := entry.Info()
if err != nil {
continue
}
if now.Sub(info.ModTime()) > maxAge {
os.Remove(filepath.Join(dir, entry.Name()))
}
}
return nil
}
// GetDiskCacheInfo 获取磁盘缓存目录信息
func GetDiskCacheInfo() (fileCount int, totalSize int64, err error) {
dir := GetDiskCacheDir()
entries, err := os.ReadDir(dir)
if err != nil {
if os.IsNotExist(err) {
return 0, 0, nil
}
return 0, 0, err
}
for _, entry := range entries {
if entry.IsDir() {
continue
}
info, err := entry.Info()
if err != nil {
continue
}
fileCount++
totalSize += info.Size()
}
return fileCount, totalSize, nil
}
// ShouldUseDiskCache 判断是否应该使用磁盘缓存
func ShouldUseDiskCache(dataSize int64) bool {
if !IsDiskCacheEnabled() {
return false
}
threshold := GetDiskCacheThresholdBytes()
if dataSize < threshold {
return false
}
return IsDiskCacheAvailable(dataSize)
}

View File

@@ -139,12 +139,29 @@ func IncrementMemoryCacheHits() {
atomic.AddInt64(&diskCacheStats.MemoryCacheHits, 1)
}
// ResetDiskCacheStats 重置统计信息(不重置当前使用量)
// ResetDiskCacheStats 重置命中统计信息(不重置当前使用量)
func ResetDiskCacheStats() {
atomic.StoreInt64(&diskCacheStats.DiskCacheHits, 0)
atomic.StoreInt64(&diskCacheStats.MemoryCacheHits, 0)
}
// ResetDiskCacheUsage 重置磁盘缓存使用量统计(用于清理缓存后)
func ResetDiskCacheUsage() {
atomic.StoreInt64(&diskCacheStats.ActiveDiskFiles, 0)
atomic.StoreInt64(&diskCacheStats.CurrentDiskUsageBytes, 0)
}
// SyncDiskCacheStats 从实际磁盘状态同步统计信息
// 用于修正统计与实际不符的情况
func SyncDiskCacheStats() {
fileCount, totalSize, err := GetDiskCacheInfo()
if err != nil {
return
}
atomic.StoreInt64(&diskCacheStats.ActiveDiskFiles, int64(fileCount))
atomic.StoreInt64(&diskCacheStats.CurrentDiskUsageBytes, totalSize)
}
// IsDiskCacheAvailable 检查是否可以创建新的磁盘缓存
func IsDiskCacheAvailable(requestSize int64) bool {
if !IsDiskCacheEnabled() {